Init api

Outfleet::log fixing

Dockerfile

@@ -5,8 +5,7 @@ WORKDIR /app
 COPY requirements.txt .
 COPY static static
 COPY templates templates
-COPY main.py .
-COPY lib.py .
+COPY *.py .
 
 RUN pip install --no-cache-dir -r requirements.txt
 

README.md

@@ -86,4 +86,6 @@ Keep in mind that all user keys are stored in a single **config.yaml** file. If
 
 ## Authors
 
-* **UltraDesu** - *Humble amateur developer* - [UltraDesu](https://github.com/house-of-vanity) - *All the work*
+* **UltraDesu** - *Humble amateur developer* - [UltraDesu](https://github.com/house-of-vanity) - *Author*
+* **Contributors**
+* * @Sanapach

k8s.py

@@ -1,24 +1,67 @@
 import base64
 import json
+import uuid
 import yaml
 import logging
-from kubernetes import client, config
+import threading
+import time
+
+import lib
+from kubernetes import client, config as kube_config
 from kubernetes.client.rest import ApiException
 
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-    datefmt="%d-%m-%Y %H:%M:%S",
-)
-
 log = logging.getLogger("OutFleet.k8s")
-file_handler = logging.FileHandler("sync.log")
-file_handler.setLevel(logging.DEBUG)
-formatter = logging.Formatter(
-    "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-)
-file_handler.setFormatter(formatter)
-log.addHandler(file_handler)
+
+NAMESPACE = False
+SERVERS = list()
+CONFIG = None
+V1 = None
+K8S_DETECTED = False
+
+
+def discovery_servers():
+    global CONFIG
+    interval = 60
+    log = logging.getLogger("OutFleet.discovery")
+
+    if K8S_DETECTED:
+        while True:
+            pods = V1.list_namespaced_pod(NAMESPACE, label_selector="app=shadowbox")
+            log.debug(f"Started discovery thread every {interval}")
+            for pod in pods.items:
+                log.debug(f"Found Outline server pod {pod.metadata.name}")
+                container_log = V1.read_namespaced_pod_log(name=pod.metadata.name, namespace=NAMESPACE, container='manager-config-json')
+                secret = json.loads(container_log.replace('\'', '\"'))
+                config = lib.get_config()
+                config_servers = find_server(secret, config["servers"])
+                #log.info(f"config_servers {config_servers}")
+                if len(config_servers) > 0:
+                    log.debug(f"Already exist")
+                    pass
+                else:
+                    with lib.lock:
+                        config["servers"][str(uuid.uuid4())] = {
+                            "cert": secret["certSha256"],
+                            "name": f"{pod.metadata.name}",
+                            "comment": f"{pod.spec.node_name}",
+                            "url": secret["apiUrl"],
+                        }
+                        write_config(config)
+                    log.info(f"Added discovered server")
+            time.sleep(interval)
+        
+        
+
+
+
+def find_server(search_data, servers):
+    found_servers = {}
+    for server_id, server_info in servers.items():
+        if server_info["url"] == search_data["apiUrl"] and server_info["cert"] == search_data["certSha256"]:
+            found_servers[server_id] = server_info
+    return found_servers
+
+
 
 def write_config(config):
     config_map = client.V1ConfigMap(
@@ -33,45 +76,53 @@ def write_config(config):
         data={"config.yaml": yaml.dump(config)}
     )
     try:
-        api_response = v1.create_namespaced_config_map(
+        api_response = V1.create_namespaced_config_map(
             namespace=NAMESPACE,
             body=config_map,
         )
     except ApiException as e:
-        api_response = v1.patch_namespaced_config_map(
+        api_response = V1.patch_namespaced_config_map(
             name="config-outfleet",
             namespace=NAMESPACE,
             body=config_map,
         )
+    log.info("Updated config in Kubernetes ConfigMap [config-outfleet]")
 
-config.load_incluster_config()
 
-v1 = client.CoreV1Api()
+def reload_config():
+    global CONFIG
+    while True:
+        new_config = yaml.safe_load(V1.read_namespaced_config_map(name="config-outfleet", namespace=NAMESPACE).data['config.yaml'])
+        with lib.lock:
+            CONFIG = new_config
+        log.debug(f"Synced system config with ConfigMap [config-outfleet].")
+        time.sleep(30)
 
-NAMESPACE = False
-SERVERS = list()
-CONFIG = None
 
-log.info("Checking for Kubernetes environment")
 try:
-    with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as f:
-        NAMESPACE = f.read().strip()
-        log.info(f"Found Kubernetes environment. Namespace {NAMESPACE}")
-except IOError:
+    kube_config.load_incluster_config()
+    V1 = client.CoreV1Api()
+    if V1 != None:
+        K8S_DETECTED = True
+    try:
+        with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as f:
+            NAMESPACE = f.read().strip()
+        log.info(f"Found Kubernetes environment. Deployed to namespace '{NAMESPACE}'")
+        try:
+            CONFIG = yaml.safe_load(V1.read_namespaced_config_map(name="config-outfleet", namespace=NAMESPACE).data['config.yaml'])
+            log.info(f"ConfigMap loaded from Kubernetes API. Servers: {len(CONFIG['servers'])}, Clients: {len(CONFIG['clients'])}. Started monitoring for changes every minute.")
+        except Exception as e:
+            try:
+                write_config({"clients": [], "servers": {}, "ui_hostname": "accessible-address.com"})
+                CONFIG = yaml.safe_load(V1.read_namespaced_config_map(name="config-outfleet", namespace=NAMESPACE).data['config.yaml'])
+                log.info("Created new ConfigMap [config-outfleet]. Started monitoring for changes every minute.")
+            except Exception as e:
+                log.info(f"Failed to create new ConfigMap [config-outfleet] {e}")
+        thread = threading.Thread(target=reload_config)
+        thread.start()
+        
+    except:
+        log.info("Kubernetes environment not detected")
+except:
     log.info("Kubernetes environment not detected")
-    pass
-
-# config = v1.list_namespaced_config_map(NAMESPACE, label_selector="app=outfleet").items["data"]["config.yaml"]
-try:
-    CONFIG = yaml.safe_load(v1.read_namespaced_config_map(name="config-outfleet", namespace=NAMESPACE).data['config.yaml'])
-    log.info(f"ConfigMap config.yaml loaded from Kubernetes API. Servers: {len(CONFIG['servers'])}, Clients: {len(CONFIG['clients'])}")
-except ApiException as e:
-    log.warning(f"ConfigMap not found. Fisrt run?")
-
-#servers = v1.list_namespaced_secret(NAMESPACE, label_selector="app=shadowbox")
-
-if not CONFIG:
-    log.info(f"Creating new ConfigMap [config-outfleet]")
-    write_config({"clients": [], "servers": [], "ui_hostname": "accessible-address.com"})
-    CONFIG = yaml.safe_load(v1.read_namespaced_config_map(name="config-outfleet", namespace=NAMESPACE).data['config.yaml'])
 

lib.py

@@ -1,5 +1,6 @@
 import argparse
 import logging
+import threading
 from typing import TypedDict, List
 from outline_vpn.outline_vpn import OutlineKey, OutlineVPN
 import yaml
@@ -21,9 +22,14 @@ parser.add_argument(
     help="Config file location",
 )
 
+lock = threading.Lock()
+
+
 args = parser.parse_args()
 def get_config():
-    if not k8s.NAMESPACE:
+    if k8s.CONFIG:
+        return k8s.CONFIG
+    else:
         try:
             with open(args.config, "r") as file:
                 config = yaml.safe_load(file)
@@ -35,18 +41,16 @@ def get_config():
                 log.error(f"Couldn't create config. {exp}")
                 return None
         return config
-    else:
-        return k8s.CONFIG
     
 def write_config(config):
-    if not k8s.NAMESPACE:
+    if k8s.CONFIG:
+        k8s.write_config(config)
+    else:
         try:
             with open(args.config, "w") as file:
                 yaml.safe_dump(config, file)
         except Exception as e:
             log.error(f"Couldn't write Outfleet config: {e}")
-    else:
-        k8s.write_config(config)
 
 
 class ServerDict(TypedDict):
@@ -95,13 +99,6 @@ class Server:
             "keys": self.client.get_keys(),
         }
         self.log = logging.getLogger(f'OutFleet.server[{self.data["name"]}]')
-        file_handler = logging.FileHandler("sync.log")
-        file_handler.setLevel(logging.DEBUG)
-        formatter = logging.Formatter(
-            "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-        )
-        file_handler.setFormatter(formatter)
-        self.log.addHandler(file_handler)
 
     def info(self) -> ServerDict:
         return self.data
@@ -110,7 +107,7 @@ class Server:
         # Looking for any users with provided name. len(result) != 1 is a problem.
         result = []
         for key in self.client.get_keys():
-            if key.key_id == name:
+            if key.name == name:
                 result.append(name)
                 self.log.info(f"check_client found client `{name}` config is correct.")
         if len(result) != 1:
@@ -121,7 +118,7 @@ class Server:
         else:
             return True
 
-    def apply_config(self, config, CFG_PATH):
+    def apply_config(self, config):
         if config.get("name"):
             self.client.set_server_name(config.get("name"))
             self.log.info(

main.py

@@ -1,16 +1,17 @@
+import threading
+import time
 import yaml
 import logging
 from datetime import datetime
 import random
 import string
-import argparse
 import uuid
 
 
 import k8s
-from flask import Flask, render_template, request, url_for, redirect
+from flask import Flask, render_template, request, url_for, redirect, jsonify
 from flask_cors import CORS
-from lib import Server, write_config, get_config, args
+from lib import Server, write_config, get_config, args, lock
 
 
 logging.getLogger("werkzeug").setLevel(logging.ERROR)
@@ -23,20 +24,18 @@ logging.basicConfig(
 
 log = logging.getLogger("OutFleet")
 file_handler = logging.FileHandler("sync.log")
-file_handler.setLevel(logging.DEBUG)
 formatter = logging.Formatter(
     "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 )
 file_handler.setFormatter(formatter)
 log.addHandler(file_handler)
 
-
 CFG_PATH = args.config
 NAMESPACE = k8s.NAMESPACE
 SERVERS = list()
 BROKEN_SERVERS = list()
 CLIENTS = dict()
-VERSION = '3'
+VERSION = '5'
 HOSTNAME = ""
 app = Flask(__name__)
 CORS(app)
@@ -53,45 +52,46 @@ def random_string(length=64):
 
 
 
-def update_state():
-    global SERVERS
-    global CLIENTS
-    global BROKEN_SERVERS
-    global HOSTNAME
-
-    SERVERS = list()
-    BROKEN_SERVERS = list()
-    CLIENTS = dict()
-    config = get_config()
-
-
-    if config:
-        HOSTNAME = config.get("ui_hostname", "my-own-SSL-ENABLED-domain.com")
-        servers = config.get("servers", dict())
-        for local_server_id, server_config in servers.items():
-            try:
-                server = Server(
-                    url=server_config["url"],
-                    cert=server_config["cert"],
-                    comment=server_config["comment"],
-                    local_server_id=local_server_id,
-                )
-                SERVERS.append(server)
-                log.info(
-                    "Server state updated: %s, [%s]",
-                    server.info()["name"],
-                    local_server_id,
-                )
-            except Exception as e:
-                BROKEN_SERVERS.append({
-                    "config": server_config,
-                    "error": e,
-                    "id": local_server_id
-                })
-                log.warning("Can't access server: %s - %s", server_config["url"], e)
-
-        CLIENTS = config.get("clients", dict())
+def update_state(timer=40):
+    
+    while True:
+        with lock:
+            global SERVERS
+            global CLIENTS
+            global BROKEN_SERVERS
+            global HOSTNAME
+            config = get_config()
 
+            if config:
+                HOSTNAME = config.get("ui_hostname", "my-own-SSL-ENABLED-domain.com")
+                servers = config.get("servers", dict())
+                _SERVERS = list()
+                for local_server_id, server_config in servers.items():
+                    try:
+                        server = Server(
+                            url=server_config["url"],
+                            cert=server_config["cert"],
+                            comment=server_config.get("comment", ''),
+                            local_server_id=local_server_id,
+                        )
+                        _SERVERS.append(server)
+                        log.debug(
+                            "Server state updated: %s, [%s]",
+                            server.info()["name"],
+                            local_server_id,
+                        )
+                    except Exception as e:
+                        BROKEN_SERVERS.append({
+                            "config": server_config,
+                            "error": e,
+                            "id": local_server_id
+                        })
+                        log.warning("Can't access server: %s - %s", server_config["url"], e)
+                SERVERS = _SERVERS
+                CLIENTS = config.get("clients", dict())
+        if timer == 0:
+            break
+        time.sleep(40)
 
 @app.route("/", methods=["GET", "POST"])
 def index():
@@ -101,6 +101,7 @@ def index():
             "index.html",
             SERVERS=SERVERS,
             VERSION=VERSION,
+            K8S_NAMESPACE=k8s.NAMESPACE,
             BROKEN_SERVERS=BROKEN_SERVERS,
             nt=request.args.get("nt"),
             nl=request.args.get("nl"),
@@ -114,8 +115,35 @@ def index():
         server = next(
             (item for item in SERVERS if item.info()["local_server_id"] == server), None
         )
-        server.apply_config(request.form, CFG_PATH)
-        update_state()
+        server.apply_config(request.form)
+        update_state(timer=0)
+        return redirect(
+            url_for(
+                "index",
+                nt="Updated Outline VPN Server",
+                selected_server=request.args.get("selected_server"),
+            )
+        )
+    else:
+        return redirect(url_for("index"))
+
+@app.route("/servers", methods=["GET", "POST"])
+@app.route("/servers/<string:local_server_id>", methods=["GET", "POST", "DELETE"])
+def servers(local_server_id=None):
+    if local_server_id:
+        log.info(f"Got {local_server_id} Config {get_config()}")
+        server = get_config()['servers'].get(local_server_id, None)
+        return jsonify(server)
+    if request.method == "GET":
+        return jsonify({ "servers": get_config()['servers']})
+
+    elif request.method == "POST":
+        server = request.form["server_id"]
+        server = next(
+            (item for item in SERVERS if item.info()["local_server_id"] == server), None
+        )
+        server.apply_config(request.form)
+        update_state(timer=0)
         return redirect(
             url_for(
                 "index",
@@ -135,6 +163,7 @@ def clients():
             SERVERS=SERVERS,
             CLIENTS=CLIENTS,
             VERSION=VERSION,
+            K8S_NAMESPACE=k8s.NAMESPACE,
             nt=request.args.get("nt"),
             nl=request.args.get("nl"),
             selected_client=request.args.get("selected_client"),
@@ -168,7 +197,7 @@ def add_server():
             config["servers"] = servers
             write_config(config)
             log.info("Added server: %s", new_server.data["name"])
-            update_state()
+            update_state(timer=0)
             return redirect(url_for("index", nt="Added Outline VPN Server"))
         except Exception as e:
             return redirect(
@@ -195,7 +224,7 @@ def del_server():
                 pass
         write_config(config)
         log.info("Deleting server %s [%s]", server_name, request.form.get("local_server_id"))
-    update_state()
+    update_state(timer=0)
     return redirect(url_for("index", nt=f"Server {server_name} has been deleted"))
 
 
@@ -260,7 +289,7 @@ def add_client():
                         request.form.get("name"),
                         server.data["name"],
                     )
-        update_state()
+        update_state(timer=0)
         return redirect(
             url_for(
                 "clients",
@@ -294,7 +323,7 @@ def del_client():
         config["clients"].pop(user_id)
         write_config(config)
         log.info("Deleting client %s", request.form.get("name"))
-    update_state()
+    update_state(timer=0)
     return redirect(url_for("clients", nt="User has been deleted"))
 
 
@@ -357,43 +386,42 @@ def sync():
             lines=lines,
         )
     if request.method == "POST":
-        log = logging.getLogger("sync")
-        file_handler = logging.FileHandler("sync.log")
-        file_handler.setLevel(logging.DEBUG)
-        formatter = logging.Formatter(
-            "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-        )
-        file_handler.setFormatter(formatter)
-        log.addHandler(file_handler)
-        if request.form.get("wipe") == 'all':
-            for server in SERVERS:
-                log.info("Wiping all keys on [%s]", server.data["name"])
-                for client in server.data['keys']:
-                    server.delete_key(client.key_id)
+        with lock:
+            if request.form.get("wipe") == 'all':
+                for server in SERVERS:
+                    log.info("Wiping all keys on [%s]", server.data["name"])
+                    for client in server.data['keys']:
+                        server.delete_key(client.key_id)
 
         server_hash = {}
-        for server in SERVERS:
-            server_hash[server.data["local_server_id"]] = server
-        for key, client in CLIENTS.items():
-            for u_server_id in client["servers"]:
-                if u_server_id in server_hash:
-                    if not server_hash[u_server_id].check_client(client["name"]):
-                        log.warning(
-                            f"Client {client['name']} absent on {server_hash[u_server_id].data['name']}"
-                        )
-                        server_hash[u_server_id].create_key(client["name"])
+        with lock:
+            for server in SERVERS:
+                server_hash[server.data["local_server_id"]] = server
+        with lock:
+            for key, client in CLIENTS.items():
+                for u_server_id in client["servers"]:
+                    if u_server_id in server_hash:
+                        if not server_hash[u_server_id].check_client(client["name"]):
+                            log.warning(
+                                f"Client {client['name']} absent on {server_hash[u_server_id].data['name']}"
+                            )
+                            server_hash[u_server_id].create_key(client["name"])
+                        else:
+                            log.info(
+                                f"Client {client['name']} already present on {server_hash[u_server_id].data['name']}"
+                            )
                     else:
                         log.info(
-                            f"Client {client['name']} already present on {server_hash[u_server_id].data['name']}"
+                            f"Client {client['name']} incorrect server_id {u_server_id}"
                         )
-                else:
-                    log.info(
-                        f"Client {client['name']} incorrect server_id {u_server_id}"
-                    )
-        update_state()
+        update_state(timer=0)
         return redirect(url_for("sync"))
 
 
 if __name__ == "__main__":
-    update_state()
+    update_state_thread = threading.Thread(target=update_state)
+    update_state_thread.start()
+
+    discovery_servers_thread = threading.Thread(target=k8s.discovery_servers)
+    discovery_servers_thread.start()
     app.run(host="0.0.0.0")