add token-based auth

app/auth.py

@@ -0,0 +1,34 @@
+import sqlite3
+
+#def __init__(self, path: str) -> None:
+#    self.db_connection = sqlite3.connect(path)
+#    cursor = self.db_connection.cursor()
+#    cursor.execute('''
+#        CREATE TABLE IF NOT EXISTS Tokens (
+#            TOKEN TEXT PRIMARY KEY NOT NULL,
+#            USER TEXT NOT NULL,
+#            NAME TEXT NOT NULL
+#        )
+#    ''')
+#
+#    cursor.execute('''
+#        CREATE TABLE IF NOT EXISTS Domains (
+#            DOMAIN TEXT PRIMARY KEY NOT NULL,
+#            USER TEXT NOT NULL
+#        )
+#    ''')
+#
+#    self.db_connection.commit()
+#    cursor.close()
+
+def check_token_domain(db_path: str, token: str, domain: str) -> bool:
+    with sqlite3.connect(db_path) as connection:
+        cursor = connection.cursor()
+        cursor.execute('''
+            SELECT COUNT(*) FROM Tokens
+            JOIN Domains ON Tokens.user = Domains.user
+            WHERE token = ?
+                AND domain = ?
+        ''', (token,domain))
+
+    return cursor.fetchone()[0] == 1

app/main.py

@@ -1,10 +1,11 @@
 from flask import Flask, jsonify, request
+import auth
 import re
 
 
 
 HOSTS_PATH = "/app/hosts.d/izbi"
-
+DB_PATH = "/app/sqlite.db"
 
 
 app = Flask(__name__)
@@ -13,23 +14,23 @@ hosts = {}
 
 
 
-def update_hosts(path: str, hosts: dict = {}): bool
+def update_hosts(path: str, hosts: dict = {}) -> bool:
     try:
         hosts_str = ""
         for domain, ip in hosts.items():
             hosts_str += f"{ip}\t{domain}\n"
         with open(path, "w") as f:
             f.write(hosts_str)
-        print(f"ddns - updated {path}")
+        print(f"ddns - updated {path}", flush=True)
         return True
     except Exception as e:
-        print(e)
+        print(e, flush=True)
         return False
 
 
 
-def validate_ip(ip: str): bool
+def validate_ip(ip: str) -> bool:
-    exp = "^((25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\.){3}(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])$"
+    exp = r"^((25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\.){3}(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])$"
     return re.search(exp, ip)
 
 
@@ -42,26 +43,38 @@ def health():
 
 @app.route('/update', methods=["GET"])
 def update_addr():
-    if 'domain' not in request.args:
+    try:
-        return jsonify({"status": "400", "code": "domain-error", "comment": "Missing or invalid domain"}), 400
+        if 'domain' not in request.args:
-    domain = request.args['domain']
+            return jsonify({"status": "400", "code": "domain-error", "comment": "Missing or invalid domain"}), 400
+        domain = request.args['domain']
 
-    if 'ip' not in request.args:
+        if 'ip' not in request.args:
-        ip = request.remote_addr
+            ip = request.remote_addr
-    elif not validate_ip(request.args['ip']):
+        elif not validate_ip(request.args['ip']):
-        return jsonify({"status": "400", "code": "ip-error", "comment": "Invalid IP"}), 400
+            return jsonify({"status": "400", "code": "ip-error", "comment": "Invalid IP"}), 400
-    else:
+        else:
-        ip = request.args['ip']
+            ip = request.args['ip']
 
-    print(f"ddns - update request of [{domain}, {ip}] from {request.remote_addr}")
+        if 'token' not in request.args:
+            return jsonify({"status": "400", "code": "token-error", "comment": "Missing or invalid token"}), 400
+        token = request.args['token']
 
-    hosts[domain] = ip
+        if not auth.check_token_domain(DB_PATH, token, domain):
-    success = update_hosts(HOSTS_PATH, hosts)
+            print(f"ddns - unauthorized update attempt from {request.remote_addr} of {domain}", flush=True)
+            return jsonify({"status": "401", "code": "unauthorized", "comment": "This token is not allowed to update this domain"}), 401
 
-    if not success:
+        success = True
-        return jsonify({"status": "500", "code": "internal-server-error", "An error occurred while processing request"}), 500
 
-    return jsonify({"status": "200", "code": "ok", "comment": "Updated successfully"})
+        if domain not in hosts or ip != hosts[domain]:
+            hosts[domain] = ip
+            success = update_hosts(HOSTS_PATH, hosts)
+
+        if not success:
+            return jsonify({"status": "500", "code": "internal-server-error", "comment": "An error occurred while processing request"}), 500
+
+        return jsonify({"status": "200", "code": "ok", "comment": "Updated successfully"})
+    except:
+        return jsonify({"status": "500", "code": "internal-server-error", "comment": "An error occurred while processing request"}), 500