From: Giorgio Ravera Date: Fri, 23 Jan 2026 18:04:18 +0000 (+0100) Subject: Created settings and log singletons X-Git-Tag: v0.0.1~32 X-Git-Url: http://git.giorgioravera.it/?a=commitdiff_plain;h=f47b85e1bf917fecb66dc83ac4693710ca8d11ad;p=network-manager.git Created settings and log singletons --- diff --git a/Dockerfile b/Dockerfile index c9e1775..8cbc78c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -16,6 +16,8 @@ WORKDIR /app COPY backend backend COPY frontend frontend COPY entrypoint.py entrypoint.py +COPY log log +COPY settings settings RUN chmod 755 entrypoint.py # ---------- STAGE 2: DISTROLESS ---------- @@ -40,6 +42,8 @@ WORKDIR /app COPY --from=builder /app/backend backend COPY --from=builder /app/frontend frontend COPY --from=builder /app/entrypoint.py entrypoint.py +COPY --from=builder /app/log log +COPY --from=builder /app/settings settings # Ensure Python sees the installed packages ENV PYTHONPATH="/usr/local/lib/python3.12/site-packages" diff --git a/backend/config.py b/backend/config.py deleted file mode 100644 index 6ebb929..0000000 --- a/backend/config.py +++ /dev/null @@ -1,45 +0,0 @@ -# backend/config.py - -# Import standard modules -import datetime -import os -import secrets -# Import local modules -from backend.utils import load_hash - -# Software Version -BASE_VERSION = "0.1.0" -DEVEL = os.getenv("DEV", "0") == "1" -if DEVEL: - timestamp = datetime.datetime.now().strftime("%Y%m%d-%H%M") - APP_VERSION = f"{BASE_VERSION}-dev-{timestamp}" -else: - APP_VERSION = BASE_VERSION - -# Base Image / Docker Image -BASEIMG_NAME = os.getenv("BASEIMG_NAME", "unknown") -BASEIMG_VERSION = os.getenv("BASEIMG_VERSION", "unknown") - -# Frontend related settings -FRONTEND_DIR = "/app/frontend" - -# Database related settings -DB_FILE = os.getenv("DB_FILE", "/data/database.db") -DB_RESET = os.getenv("DB_RESET", "0") == "1" - -# Hosts related settings -DOMAIN = os.getenv("DOMAIN", "example.com") -PUBLIC_IP = os.getenv("PUBLIC_IP", "127.0.0.1") - -# Web server related settings -HTTP_PORT = os.getenv("HTTP_PORT", "8000") -SECRET_KEY = os.getenv("SESSION_SECRET") -if not SECRET_KEY: - SECRET_KEY = load_hash("SECRET_KEY_FILE") or secrets.token_urlsafe(64) -LOGIN_MAX_ATTEMPTS = int(os.getenv("LOGIN_MAX_ATTEMPTS", "5")) -LOGIN_WINDOW_SECONDS = int(os.getenv("LOGIN_WINDOW_SECONDS", "600")) - -# User related settings -ADMIN_USER = os.getenv("ADMIN_USER", "admin") -ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin") -ADMIN_HASH = os.getenv("ADMIN_HASH") or load_hash("ADMIN_HASH_FILE") diff --git a/backend/db/db.py b/backend/db/db.py index ac5e392..d818501 100644 --- a/backend/db/db.py +++ b/backend/db/db.py @@ -3,8 +3,10 @@ # Import standard modules import os import sqlite3 -# Import local modules -from backend.config import DB_FILE +# Import Settings +from settings.settings import settings +# Import Log +from log.log import get_logger _connection = None _init_functions = [] @@ -19,8 +21,8 @@ def register_init(func): def get_db(): global _connection if _connection is None: - os.makedirs(os.path.dirname(DB_FILE) or ".", exist_ok=True) - _connection = sqlite3.connect(DB_FILE, check_same_thread=False) + os.makedirs(os.path.dirname(settings.DB_FILE) or ".", exist_ok=True) + _connection = sqlite3.connect(settings.DB_FILE, check_same_thread=False) _connection.row_factory = sqlite3.Row _connection.execute("PRAGMA foreign_keys = ON;") return _connection @@ -29,7 +31,8 @@ def get_db(): # Init Database # ----------------------------- def init_db(): - print(f"INFO: Starting DB Initialization.") + logger = get_logger(__name__) + logger.info("Starting DB Initialization") conn = get_db() cur = conn.cursor() @@ -40,4 +43,4 @@ def init_db(): conn.commit() conn.close() - print(f"INFO: DB Initialization Completed.") \ No newline at end of file + logger.info("DB Initialization Completed") diff --git a/backend/db/hosts.py b/backend/db/hosts.py index 9f0e98f..aacc09d 100644 --- a/backend/db/hosts.py +++ b/backend/db/hosts.py @@ -2,10 +2,14 @@ # Import standard modules import ipaddress +import logging import os # Import local modules -from backend.db.db import get_db -from backend.db.db import register_init +from backend.db.db import get_db, register_init +# Import Settings +from settings.settings import settings +# Import Log +from log.log import get_logger # ----------------------------- # SELECT ALL HOSTS @@ -79,20 +83,19 @@ def delete_host(host_id: int): # ----------------------------- @register_init def init_db_hosts_table(cur): - from backend.config import DOMAIN - from backend.config import PUBLIC_IP + logger = get_logger(__name__) - # GLOBAL SETTINGS + # SETTINGS TABLE cur.execute(""" CREATE TABLE settings ( key TEXT PRIMARY KEY, value TEXT ); """) - cur.execute("INSERT INTO settings (key, value) VALUES (?, ?)", ("domain", DOMAIN)) - cur.execute("INSERT INTO settings (key, value) VALUES (?, ?)", ("external_ipv4", PUBLIC_IP)) + cur.execute("INSERT INTO settings (key, value) VALUES (?, ?)", ("domain", settings.DOMAIN)) + cur.execute("INSERT INTO settings (key, value) VALUES (?, ?)", ("external_ipv4", settings.PUBLIC_IP)) - # HOSTS + # HOSTS TABLE cur.execute(""" CREATE TABLE hosts ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -106,7 +109,7 @@ def init_db_hosts_table(cur): """) cur.execute("CREATE INDEX idx_hosts_name ON hosts(name);") - # ALIASES + # ALIASES TABLE cur.execute(""" CREATE TABLE aliases ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -119,7 +122,7 @@ def init_db_hosts_table(cur): """) cur.execute("CREATE INDEX idx_aliases_host ON aliases(host_id);") - # TXT RECORDS + # TXT TABLE cur.execute(""" CREATE TABLE txt_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -132,5 +135,5 @@ def init_db_hosts_table(cur): """) cur.execute("CREATE INDEX idx_txt_host ON txt_records(host_id);") - print(f"INFO: - HOSTS DB: Database initialized successfully for {DOMAIN}.") - print(f"INFO: - HOSTS DB: Public IP: {PUBLIC_IP}.") + logger.info("HOSTS DB: Database initialized successfully for %s", settings.DOMAIN) + logger.info("HOSTS DB: Public IP: %s", settings.PUBLIC_IP) diff --git a/backend/db/users.py b/backend/db/users.py index 48a427c..c3ff7ff 100644 --- a/backend/db/users.py +++ b/backend/db/users.py @@ -1,10 +1,16 @@ # backend/db/users.py # Import standard modules +import bcrypt import json +import logging import os # Import local modules from backend.db.db import get_db, register_init +# Import Settings +from settings.settings import settings +# Import Log +from log.log import get_logger # ================================ # Create hash password @@ -28,10 +34,9 @@ def get_user_by_username(username): # ----------------------------- @register_init def init_db_users_table(cur): - from backend.config import ADMIN_USER - from backend.config import ADMIN_PASSWORD - from backend.config import ADMIN_HASH + logger = get_logger(__name__) + # USERS TABLE cur.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -52,25 +57,26 @@ def init_db_users_table(cur): """) cur.execute("CREATE INDEX idx_users_username ON users(username);") # Insert default admin user - if not ADMIN_HASH: - ADMIN_HASH = hash_password(ADMIN_PASSWORD) + if not settings.ADMIN_PASSWORD_HASH: + settings.ADMIN_PASSWORD_HASH = hash_password(settings.ADMIN_PASSWORD) else: - ADMIN_PASSWORD = "(hidden)" + settings.ADMIN_PASSWORD = "(hidden)" cur.execute(""" INSERT INTO users ( username, password_hash, email, is_admin, modules, status, created_at, updated_at, password_changed_at ) VALUES (?, ?, ?, ?, ?, ?, strftime('%s','now'), strftime('%s','now'), strftime('%s','now')); """, ( - ADMIN_USER, - ADMIN_HASH, + settings.ADMIN_USER, + settings.ADMIN_PASSWORD_HASH, "admin@example.com", 1, '["dns","dhcp"]', "active" )) - print(f"INFO: - USERS DB: Admin user: {ADMIN_USER} with password {ADMIN_PASSWORD} - {ADMIN_HASH}.") + logger.info("USERS DB: Admin user: %s with password %s - %s" , + settings.ADMIN_USER, settings.ADMIN_PASSWORD, settings.ADMIN_PASSWORD_HASH) # ----------------------------- # Create User diff --git a/backend/main.py b/backend/main.py index 4cecd14..7303938 100644 --- a/backend/main.py +++ b/backend/main.py @@ -1,39 +1,96 @@ # backend/main.py # import standard modules -from fastapi import FastAPI, Request, Response +from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import FileResponse, RedirectResponse, JSONResponse +from fastapi.responses import FileResponse, RedirectResponse, JSONResponse, Response +import logging import os -# Import local modules -from backend.security import is_logged_in, apply_session +# Import Routers +from backend.routes.about import router as about_router from backend.routes.health import router as health_router from backend.routes.login import router as login_router from backend.routes.hosts import router as hosts_router -# Import config variables -from backend.config import FRONTEND_DIR, HTTP_PORT +# Import Security +from backend.security import is_logged_in, apply_session +# Import Settings +from settings.settings import settings +# Import Logging +from log.log import setup_logging, get_logger + +# ------------------------------------------------------------------------------ +# Logging setup +# ------------------------------------------------------------------------------ +setup_logging(settings.LOG_LEVEL, settings.LOG_TO_FILE, settings.LOG_FILE) +logger = get_logger(__name__) + +# ------------------------------------------------------------------------------ +# App init +# ------------------------------------------------------------------------------ +app = FastAPI( + title=settings.APP_NAME, + version=settings.APP_VERSION, +) -# Start FastAPI app -app = FastAPI() +# ------------------------------------------------------------------------------ +# Routers +# ------------------------------------------------------------------------------ +app.include_router(about_router) app.include_router(health_router) app.include_router(login_router) app.include_router(hosts_router) -# Allow frontend JS to call the API +# ------------------------------------------------------------------------------ +# Startup log +# ------------------------------------------------------------------------------ +@app.on_event("startup") +async def startup_log(): + logger = get_logger(__name__) + + safe_secret = "****" if settings.SECRET_KEY else "undefined" + safe_admin_pwd = "****" if settings.ADMIN_PASSWORD else "undefined" + safe_admin_hash = "****" if settings.ADMIN_PASSWORD_HASH else "undefined" + + logger.info( + "%s starting | app_version=%s | baseimg_version=%s", + settings.APP_NAME, settings.APP_VERSION, settings.BASEIMG_VERSION + ) + logger.info( + "App settings: frontend=%s | port=%d | secret=%s", + settings.FRONTEND_DIR, settings.HTTP_PORT, safe_secret + ) + logger.info( + "Database: file=%s | reset=%s", + settings.DB_FILE, settings.DB_RESET + ) + logger.info( + "Log: level=%s, to_file=%s, file=%s", + settings.LOG_LEVEL, settings.LOG_TO_FILE, settings.LOG_FILE + ) + logger.info( + "Users: admin=%s | password=%s | hash=%s | hash_file=%s", + settings.ADMIN_USER, safe_admin_pwd, safe_admin_hash, settings.ADMIN_PASSWORD_HASH_FILE + ) + +# ------------------------------------------------------------------------------ +# CORS +# ------------------------------------------------------------------------------ +cors_origins = [ + f"http://localhost:{settings.HTTP_PORT}", + f"http://127.0.0.1:{settings.HTTP_PORT}", +] + app.add_middleware( CORSMiddleware, - allow_origins=[ - f"http://localhost:{HTTP_PORT}", - f"http://127.0.0.1:{HTTP_PORT}", - ], - allow_methods=["GET", "POST", "PUT", "DELETE"], - allow_headers=["Content-Type"], + allow_origins=cors_origins, + allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], + allow_headers=["Content-Type", "Authorization"], + allow_credentials=True, ) -# --------------------------------------------------------- -# Middleware to manage Login -# --------------------------------------------------------- - +# ------------------------------------------------------------------------------ +# Session / Auth middleware +# ------------------------------------------------------------------------------ @app.middleware("http") async def session_middleware(request: Request, call_next): path = request.url.path @@ -47,11 +104,8 @@ async def session_middleware(request: Request, call_next): if ( path.startswith("/css") or path.startswith("/js") or - path.endswith(".js") or - path.endswith(".css") or - path.endswith(".png") or - path.endswith(".jpg") or - path.endswith(".ico") + path.startswith("/static") or + path.endswith((".js", ".css", ".png", ".jpg", ".jpeg", ".ico", ".svg", ".map")) ): return await call_next(request) @@ -59,7 +113,6 @@ async def session_middleware(request: Request, call_next): if path.startswith("/api"): if not is_logged_in(request): return JSONResponse({"error": "Not authenticated"}, status_code=401) - response = await call_next(request) # Sliding expiration apply_session(response, username=None, token=token) @@ -74,21 +127,21 @@ async def session_middleware(request: Request, call_next): apply_session(response, username=None, token=token) return response -# --------------------------------------------------------- -# FRONTEND PATHS (absolute paths inside Docker) -# --------------------------------------------------------- - +# ------------------------------------------------------------------------------ +# FRONTEND (FileResponse): pages and assets +# ------------------------------------------------------------------------------ # Homepage @app.get("/") def home(request: Request): - return FileResponse(os.path.join(FRONTEND_DIR, "hosts.html")) + return FileResponse(os.path.join(settings.FRONTEND_DIR, "hosts.html")) # CSS variables @app.get("/css/variables.css") -def home(request: Request): - return FileResponse(os.path.join(FRONTEND_DIR, "css/variables.css")) +def css_variables(request: Request): + return FileResponse(os.path.join(settings.FRONTEND_DIR, "css/variables.css")) -# CSS Layoyt +# CSS Layout @app.get("/css/layout.css") -def home(request: Request): - return FileResponse(os.path.join(FRONTEND_DIR, "css/layout.css")) +def css_layout(request: Request): + return FileResponse(os.path.join(settings.FRONTEND_DIR, "css/layout.css")) + diff --git a/backend/routes/about.py b/backend/routes/about.py new file mode 100644 index 0000000..096a7c3 --- /dev/null +++ b/backend/routes/about.py @@ -0,0 +1,26 @@ +# backend/about.py + +# Import standard modules +from fastapi import APIRouter +# Import Settings +from settings.settings import settings + +# Create Router +router = APIRouter() + +# --------------------------------------------------------- +# API ENDPOINTS +# --------------------------------------------------------- +@router.get("/about") +def about(): + return { + "app": { + "version": settings.APP_VERSION + }, + "baseimg": { + "name": settings.BASEIMG_NAME, + "version": settings.BASEIMG_VERSION + }, + "domain": settings.DOMAIN, + "admin_hash_loaded": settings.ADMIN_PASSWORD_HASH is not None, + } diff --git a/backend/routes/health.py b/backend/routes/health.py index 528c206..bd3e766 100644 --- a/backend/routes/health.py +++ b/backend/routes/health.py @@ -5,8 +5,8 @@ from fastapi import APIRouter import sqlite3 import time import os -# Import config variables -from backend.config import APP_VERSION, BASEIMG_NAME, BASEIMG_VERSION, DB_FILE +# Import Settings +from settings.settings import settings # Create Router router = APIRouter() @@ -14,7 +14,6 @@ router = APIRouter() # --------------------------------------------------------- # API ENDPOINTS # --------------------------------------------------------- - @router.get("/api/health", tags=["health"]) def health(): start = time.time() @@ -25,7 +24,7 @@ def health(): db_size = None try: - conn = sqlite3.connect(DB_FILE) + conn = sqlite3.connect(settings.DB_FILE) cursor = conn.cursor() cursor.execute("select sqlite_version()") @@ -36,7 +35,7 @@ def health(): conn.close() - db_size = round(os.path.getsize(DB_FILE) / (1024 * 1024), 2) + db_size = round(os.path.getsize(settings.DB_FILE) / (1024 * 1024), 2) except Exception as e: db_status = "error" @@ -46,13 +45,6 @@ def health(): return { "status": "ok" if db_status == "ok" else "degraded", - "app": { - "version": APP_VERSION - }, - "baseimg": { - "name": BASEIMG_NAME, - "version": BASEIMG_VERSION - }, "latency_ms": latency, "database": { "status": db_status, @@ -61,4 +53,3 @@ def health(): "size_mb": db_size } } - diff --git a/backend/routes/hosts.py b/backend/routes/hosts.py index caccc3a..4337955 100644 --- a/backend/routes/hosts.py +++ b/backend/routes/hosts.py @@ -13,8 +13,8 @@ from backend.db.hosts import ( update_host, delete_host ) -# Import config variables -from backend.config import FRONTEND_DIR +# Import Settings +from settings.settings import settings # Create Router router = APIRouter() @@ -22,26 +22,24 @@ router = APIRouter() # --------------------------------------------------------- # FRONTEND PATHS (absolute paths inside Docker) # --------------------------------------------------------- - # Hosts page @router.get("/hosts") def hosts(request: Request): - return FileResponse(os.path.join(FRONTEND_DIR, "hosts.html")) + return FileResponse(os.path.join(settings.FRONTEND_DIR, "hosts.html")) # Serve hosts.css @router.get("/css/hosts.css") def css_hosts(): - return FileResponse(os.path.join(FRONTEND_DIR, "css/hosts.css")) + return FileResponse(os.path.join(settings.FRONTEND_DIR, "css/hosts.css")) # Serve hosts.js @router.get("/js/hosts.js") def css_hosts(): - return FileResponse(os.path.join(FRONTEND_DIR, "js/hosts.js")) + return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/hosts.js")) # --------------------------------------------------------- # API ENDPOINTS # --------------------------------------------------------- - @router.get("/api/hosts") def api_get_hosts(request: Request): return get_hosts() diff --git a/backend/routes/login.py b/backend/routes/login.py index 91a43db..5a37034 100644 --- a/backend/routes/login.py +++ b/backend/routes/login.py @@ -7,8 +7,8 @@ import os import time # Import local modules from backend.security import verify_login, apply_session -# Import config variables -from backend.config import FRONTEND_DIR, LOGIN_MAX_ATTEMPTS, LOGIN_WINDOW_SECONDS +# Import Settings +from settings.settings import settings # Create Router router = APIRouter() @@ -20,9 +20,9 @@ def check_rate_limit(ip: str): now = time.time() attempts = login_attempts.get(ip, []) # tieni solo tentativi negli ultimi LOGIN_WINDOW_SECONDS secondi - attempts = [t for t in attempts if now - t < LOGIN_WINDOW_SECONDS] + attempts = [t for t in attempts if now - t < settings.LOGIN_WINDOW_SECONDS] - if len(attempts) >= LOGIN_MAX_ATTEMPTS: + if len(attempts) >= settings.LOGIN_MAX_ATTEMPTS: raise HTTPException(status_code=429, detail="Too many login attempts") # registra nuovo tentativo @@ -32,31 +32,29 @@ def check_rate_limit(ip: str): # --------------------------------------------------------- # FRONTEND PATHS (absolute paths inside Docker) # --------------------------------------------------------- - # Login page @router.get("/login") def login_page(request: Request): - return FileResponse(os.path.join(FRONTEND_DIR, "login.html")) + return FileResponse(os.path.join(settings.FRONTEND_DIR, "login.html")) # Serve login.css @router.get("/css/login.css") def css_login(): - return FileResponse(os.path.join(FRONTEND_DIR, "css/login.css")) + return FileResponse(os.path.join(settings.FRONTEND_DIR, "css/login.css")) # Serve login.js @router.get("/js/login.js") def css_login(): - return FileResponse(os.path.join(FRONTEND_DIR, "js/login.js")) + return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/login.js")) # Serve session.js @router.get("/js/session.js") def css_login(): - return FileResponse(os.path.join(FRONTEND_DIR, "js/session.js")) + return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/session.js")) # --------------------------------------------------------- # API ENDPOINTS # --------------------------------------------------------- - @router.post("/api/login") def api_login(request: Request, data: dict, response: Response): ip = request.client.host diff --git a/backend/security.py b/backend/security.py index 504d9ac..0432584 100644 --- a/backend/security.py +++ b/backend/security.py @@ -7,48 +7,51 @@ from fastapi import Request, HTTPException from itsdangerous import TimestampSigner # Import local modules from backend.db.users import get_user_by_username -from backend.utils import log_event -# Import config variables -from backend.config import FRONTEND_DIR, SECRET_KEY +# Import Settings +from settings.settings import settings +# Import Log +from log.log import get_logger -signer = TimestampSigner(SECRET_KEY) +signer = TimestampSigner(settings.SECRET_KEY) # ----------------------------- # Verify Login # ----------------------------- def verify_login(username, password): + logger = get_logger(__name__) user = get_user_by_username(username) if not user: - log_event("LOGIN failed - user not found", user=username) + logger.error("LOGIN failed - user %s not found", username) return False if user["status"] != "active": - log_event("LOGIN Failed - user disabled", user=username) + logger.error("LOGIN Failed - user %s disabled", username) return False if not bcrypt.checkpw(password.encode(), user["password_hash"].encode()): - log_event("LOGIN Failed - password wrong", user=username) + logger.error("LOGIN Failed - password wrong for user %s", username) return False - log_event("LOGIN", user=username) + logger.info("LOGIN user %s", username) return True # ---------------------------- # creates or renew the cookie # ---------------------------- def apply_session(response, username: str | None = None, token: str | None = None): + logger = get_logger(__name__) # First Login if username is not None and token is None: token = signer.sign(username).decode() - log_event("SESSION_CREATE", user=username) + logger.info("SESSION_CREATE - %s", username) if username is None: username = signer.unsign(token, max_age=86400).decode() - log_event("SESSION_UPDATE", user=username) + logger.info("SESSION_UPDATE - %s", username) if username is None or token is None: - log_event("SESSION_ERROR") + logger.error("SESSION_ERROR") return response.set_cookie( diff --git a/backend/utils.py b/backend/utils.py index b5e4cf8..ced178b 100644 --- a/backend/utils.py +++ b/backend/utils.py @@ -1,24 +1,13 @@ # backend/db/utils.py # Import standard modules -from datetime import datetime import os # ----------------------------- # Load hash from file # ----------------------------- -def load_hash(hash_file: str): - path = os.environ.get(hash_file) +def load_hash(path: str): if path and os.path.exists(path): with open(path, "r") as f: return f.read().strip() return None - -# ----------------------------- -# Log Event -# ----------------------------- -def log_event(event: str, **fields): - ts = datetime.utcnow().isoformat() + "Z" - parts = " ".join(f"{k}={v}" for k, v in fields.items()) - print(f"INFO: {ts} {event} {parts}") - diff --git a/entrypoint.py b/entrypoint.py index 5213628..5b25579 100755 --- a/entrypoint.py +++ b/entrypoint.py @@ -1,80 +1,92 @@ #!/usr/local/bin/python3 # Import standard modules +import logging import os import sys +import argparse # Import local modules from backend.db.db import init_db import backend.db.hosts import backend.db.users +# Import Settings +from settings.settings import settings +# Import Log +from log.log import setup_logging, get_logger # ================================ -# Variables +# Parse CLI arguments # ================================ -BASEIMG_NAME = "network-manager-distroless" -BASEIMG_VERSION = "0.2" - -from backend.config import DB_FILE -from backend.config import DB_RESET +def parse_args(): + parser = argparse.ArgumentParser(add_help=False) + parser.add_argument("--reset", action="store_true") + parser.add_argument("--domain") + parser.add_argument("--public-ip") + parser.add_argument("cmd", nargs=argparse.REMAINDER) + return parser.parse_args() # ================================ # Create DB if needed # ================================ -def docker_create_db(): +def docker_create_db(logger): # Reset database if requested - if DB_RESET and os.path.exists(DB_FILE): - print("INFO: Removing existing database.") - os.remove(DB_FILE) + if settings.DB_RESET and os.path.exists(settings.DB_FILE): + logger.info("Removing existing database: %s", settings.DB_FILE) + os.remove(settings.DB_FILE) # Skip creation if DB already exists - if os.path.exists(DB_FILE): - print("INFO: Database already exists. Nothing to do.") + if os.path.exists(settings.DB_FILE): + logger.info("Database already exists. Nothing to do.") return - print(f"INFO: Creating database: {DB_FILE}.") + logger.info("Creating database: %s", settings.DB_FILE) # Ensure directory exists - os.makedirs(os.path.dirname(DB_FILE) or ".", exist_ok=True) + os.makedirs(os.path.dirname(settings.DB_FILE) or ".", exist_ok=True) - # Initialize all registered DB tables + # Initialize DB tables init_db() # ================================ # Entry Point # ================================ +def main(): + # Enable logging + setup_logging() + logger = get_logger("baseimg") + + # Log startup docker image + logger.info("Starting docker image %s version %s", settings.BASEIMG_NAME, settings.BASEIMG_VERSION) + + # Parse arguments + args = parse_args() + + # Apply arguments into settings + if args.reset: + settings.DB_RESET = True + if args.domain: + settings.DOMAIN = args.domain + if args.public_ip: + settings.PUBLIC_IP = args.public_ip + + # Create or update database + docker_create_db(logger) + + # If no command provided -> error + if not args.cmd: + logger.error("No command provided. Exiting.") + sys.exit(1) + + cmd = args.cmd[0] + rest = args.cmd[1:] + + logger.info("Docker image initialization completed — executing: %s %s", cmd, " ".join(rest)) + + try: + os.execvp(cmd, [cmd, *rest]) + except FileNotFoundError: + logger.critical("Command not found: %s", cmd) + sys.exit(1) -# Force flush -sys.stdout.reconfigure(line_buffering=True) - -print(f"INFO: Starting {BASEIMG_NAME} docker image version {BASEIMG_VERSION}.") -os.environ["BASEIMG_NAME"] = BASEIMG_NAME -os.environ["BASEIMG_VERSION"] = BASEIMG_VERSION - -# Parse arguments -args = sys.argv[1:] -i = 0 -while i < len(args): - if args[i] == "--reset": - DB_RESET = True - i += 1 - elif args[i] == "--domain" and i + 1 < len(args): - DOMAIN = args[i + 1] - i += 2 - elif args[i] == "--public-ip" and i + 1 < len(args): - PUBLIC_IP = args[i + 1] - i += 2 - elif args[i] == "--": - args = args[i + 1:] - break - else: - break - -# Create DB -docker_create_db() - -# Continue to CMD -if not args: - print("ERROR: No command provided to exec.") - sys.exit(1) - -os.execvp(args[0], args) +if __name__ == "__main__": + main() diff --git a/log/__init__.py b/log/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/log/log.py b/log/log.py new file mode 100644 index 0000000..155f5cb --- /dev/null +++ b/log/log.py @@ -0,0 +1,133 @@ +# log/log.py + +import logging +import logging.config +import os +import sys + +# Module-level flag to prevent re-initialization +_INITIALIZED = False + +# --------------------------------------------------------- +# Build a full dictConfig for logging +# --------------------------------------------------------- +def build_log_config(level: str = "INFO", to_file: bool = False, file: str | None = None) -> dict: + """ + Returns a complete dictConfig for the logging system, including formatters, + console/file handlers, and specific loggers (uvicorn, fastapi, etc). + """ + level = (level or "INFO").upper() + + formatters = { + "detailed": { + "format": "%(asctime)s %(levelname)s [%(name)s] %(message)s", + "datefmt": "%Y-%m-%dT%H:%M:%S%z", + }, + "access": { + "format": '%(asctime)s %(levelname)s [%(name)s] ' + '%(client_addr)s - "%(request_line)s" %(status_code)s', + "datefmt": "%Y-%m-%dT%H:%M:%S%z", + }, + } + + handlers = { + "console": { + "class": "logging.StreamHandler", + "level": level, + "formatter": "detailed", + "stream": "ext://sys.stdout", + } + } + + # Select active handler based on console + active_handlers = ["console"] + + if to_file: + if file is not None: + # Ensure the log directory exists and add a rotating file handler + log_dir = os.path.dirname(file) or "." + os.makedirs(log_dir, exist_ok=True) + handlers["file"] = { + "class": "logging.handlers.RotatingFileHandler", + "level": level, + "formatter": "detailed", + "filename": file, + "maxBytes": 5 * 1024 * 1024, + "backupCount": 5, + "encoding": "utf-8", + } + # Add active handler based on file + active_handlers.append("file") + + return { + "version": 1, + "disable_existing_loggers": False, + "formatters": formatters, + "handlers": handlers, + "root": { + "level": level, + "handlers": active_handlers, + }, + # Modules + "loggers": { + # Uvicorn core + "uvicorn": { + "level": level, + "handlers": active_handlers, + "propagate": False, + }, + # Uvicorn internal error + "uvicorn.error": { + "level": level, + "handlers": active_handlers, + "propagate": False, + }, + # Uvicorn HTTP access + "uvicorn.access": { + "level": level, + "handlers": active_handlers, + "propagate": False, + }, + # FastAPI + "fastapi": { + "level": level, + "handlers": active_handlers, + "propagate": False, + }, + }, + } + +# --------------------------------------------------------- +# Initialize logging once (singleton guard) +# --------------------------------------------------------- +def setup_logging(level: str = "INFO", to_file: bool = False, file: str | None = None) -> None: + """ + Initializes the logging system only once. Subsequent calls are no-ops. + Useful to prevent duplicated handlers or reconfiguration side effects. + """ + global _INITIALIZED + + if _INITIALIZED: + return + + config = build_log_config(level=level, to_file=to_file, file=file) + logging.config.dictConfig(config) + + logging.getLogger(__name__).info( + "Logging configured (level=%s, to_file=%s, file=%s)", + level.upper(), to_file, file + ) + + _INITIALIZED = True + +# --------------------------------------------------------- +# Get a configured logger for the given module/name +# --------------------------------------------------------- +def get_logger(name: str = None) -> logging.Logger: + """ + Returns a logger instance configured via the module setup. If setup was not + called yet, it falls back to the standard logging defaults. + """ + if not name: + name = __name__ + return logging.getLogger(name) diff --git a/settings/__init__.py b/settings/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/settings/config.py b/settings/config.py new file mode 100644 index 0000000..c0bc265 --- /dev/null +++ b/settings/config.py @@ -0,0 +1,13 @@ +# backend/config.py + +# --------------------------------------------------------- +# BASEIMG +# --------------------------------------------------------- +BASEIMG_NAME = "network-manager-distroless" +BASEIMG_VERSION = "0.3" + +# --------------------------------------------------------- +# APP +# --------------------------------------------------------- +APP_NAME = "network-manager" +APP_VERSION = "0.2.0" diff --git a/settings/default.py b/settings/default.py new file mode 100644 index 0000000..9126c0f --- /dev/null +++ b/settings/default.py @@ -0,0 +1,39 @@ +# backend/default.py + +# --------------------------------------------------------- +# Frontend +# --------------------------------------------------------- +FRONTEND_DIR = "/app/frontend" + +# --------------------------------------------------------- +# Database +# --------------------------------------------------------- +DB_FILE = "/data/database.db" +DB_RESET = False + +# --------------------------------------------------------- +# Log +# --------------------------------------------------------- +LOG_LEVEL = "INFO" +LOG_TO_FILE = False +LOG_FILE = "/data/app.log" + +# --------------------------------------------------------- +# Host +# --------------------------------------------------------- +DOMAIN = "example.com" +PUBLIC_IP = "127.0.0.1" + +# --------------------------------------------------------- +# Web +# --------------------------------------------------------- +HTTP_PORT = "8000" +LOGIN_MAX_ATTEMPTS = "5" +LOGIN_WINDOW_SECONDS = "600" + +# --------------------------------------------------------- +# Admin +# --------------------------------------------------------- +ADMIN_USER = "admin" +ADMIN_PASSWORD = "admin" +ADMIN_PASSWORD_HASH_FILE = "/run/secrets/admin_password_hash" diff --git a/settings/settings.py b/settings/settings.py new file mode 100644 index 0000000..4da9a99 --- /dev/null +++ b/settings/settings.py @@ -0,0 +1,96 @@ +# backend/settings.py + +from __future__ import annotations + +# import standard modules +import os +import secrets +import datetime +from pathlib import Path +from typing import Optional +from pydantic import BaseModel, Field, field_validator +# Import Parameters +from . import config, default + +# --------------------------------------------------------- +# Convert value to boolean +# --------------------------------------------------------- +def _to_bool(val) -> bool: + if isinstance(val, bool): + return val + if val is None: + return False + return str(val).strip().lower() in {"1", "true", "yes", "on"} + +# --------------------------------------------------------- +# Read text from file if it exists +# --------------------------------------------------------- +def _read_text_if_exists(path: Optional[str]) -> Optional[str]: + if not path: + return None + p = Path(path) + if p.exists() and p.is_file(): + try: + return p.read_text(encoding="utf-8").strip() + except Exception: + return None + return None + +# --------------------------------------------------------- +# Settings Model +# --------------------------------------------------------- +class Settings(BaseModel): + # Naming + APP_NAME: str = Field(default_factory=lambda: config.APP_NAME) + + # Versioning + APP_VERSION: str = Field(default_factory=lambda: config.APP_VERSION) + DEVEL: bool = Field(default_factory=lambda: _to_bool(os.getenv("DEV", False))) + + # Base Image / Docker Image + BASEIMG_NAME: str = Field(default_factory=lambda: config.BASEIMG_NAME) + BASEIMG_VERSION: str = Field(default_factory=lambda: config.BASEIMG_VERSION) + + # Frontend + FRONTEND_DIR: str = Field(default_factory=lambda: os.getenv("FRONTEND_DIR", default.FRONTEND_DIR)) + + # Database + DB_FILE: str = Field(default_factory=lambda: os.getenv("DB_FILE", default.DB_FILE)) + DB_RESET: bool = Field(default_factory=lambda: _to_bool(os.getenv("DB_RESET", default.DB_RESET))) + + # Log + LOG_LEVEL: str = Field(default_factory=lambda: os.getenv("LOG_LEVEL", default.LOG_LEVEL)) + LOG_TO_FILE: bool = Field(default_factory=lambda: _to_bool(os.getenv("LOG_TO_FILE", default.LOG_TO_FILE))) + LOG_FILE: str = Field(default_factory=lambda: os.getenv("LOG_FILE", default.LOG_FILE)) + + # Hosts + DOMAIN: str = Field(default_factory=lambda: os.getenv("DOMAIN", default.DOMAIN)) + PUBLIC_IP: str = Field(default_factory=lambda: os.getenv("PUBLIC_IP", default.DOMAIN)) + + # Web + HTTP_PORT: int = Field(default_factory=lambda: int(os.getenv("HTTP_PORT", default.HTTP_PORT))) + SECRET_KEY: str = Field(default_factory=lambda: ( + (os.getenv("SESSION_SECRET") or _read_text_if_exists(os.getenv("SECRET_KEY_FILE")) or secrets.token_urlsafe(64)).strip() + )) + LOGIN_MAX_ATTEMPTS: int = Field(default_factory=lambda: int(os.getenv("LOGIN_MAX_ATTEMPTS", default.LOGIN_MAX_ATTEMPTS))) + LOGIN_WINDOW_SECONDS: int = Field(default_factory=lambda: int(os.getenv("LOGIN_WINDOW_SECONDS", default.LOGIN_WINDOW_SECONDS))) + + # Admin + ADMIN_USER: str = Field(default_factory=lambda: os.getenv("ADMIN_USER", default.ADMIN_USER)) + ADMIN_PASSWORD: str = Field(default_factory=lambda: os.getenv("ADMIN_PASSWORD", default.ADMIN_PASSWORD)) + ADMIN_PASSWORD_HASH_FILE: str = Field(default_factory=lambda: os.getenv("ADMIN_PASSWORD_HASH_FILE", default.ADMIN_PASSWORD_HASH_FILE)) + ADMIN_PASSWORD_HASH: Optional[str] = Field(default_factory=lambda: ( + (os.getenv("ADMIN_PASSWORD_HASH") or _read_text_if_exists(os.getenv("ADMIN_PASSWORD_HASH_FILE", default.ADMIN_PASSWORD_HASH_FILE)) or None) + )) + + def model_post_init(self, __context) -> None: + if self.DEVEL: + ts = datetime.datetime.now().strftime("%Y%m%d-%H%M") + object.__setattr__(self, "APP_VERSION", f"{self.APP_VERSION}-dev-{ts}") + else: + object.__setattr__(self, "APP_VERSION", self.APP_VERSION) + +# --------------------------------------------------------- +# Singleton +# --------------------------------------------------------- +settings = Settings()