fix

2026-03-10 15:34:22 +00:00
parent 7635caa71d
commit d641e181ba
17 changed files with 347 additions and 7 deletions
--- a/app/main.py
+++ b/app/main.py
@@ -3,10 +3,42 @@ import os
 from datetime import datetime, timedelta, timezone

 from flask import Flask, redirect, render_template, request, url_for
+from flask_login import LoginManager, current_user, login_required, login_user, logout_user

 from app import models


+def _is_safe_redirect_url(url: str | None) -> bool:
+    """Check that redirect URL is relative to our app (prevents open redirect)."""
+    if not url:
+        return False
+    return url.startswith("/") and "//" not in url
+
+
+class User:
+    """Flask-Login compatible user wrapper."""
+
+    def __init__(self, user_dict: dict):
+        self.id = user_dict["id"]
+        self.username = user_dict["username"]
+        self.is_admin = bool(user_dict.get("is_admin", 0))
+
+    def get_id(self) -> str:
+        return str(self.id)
+
+    @property
+    def is_authenticated(self) -> bool:
+        return True
+
+    @property
+    def is_active(self) -> bool:
+        return True
+
+    @property
+    def is_anonymous(self) -> bool:
+        return False
+
+
 def _parse_report_dates(from_ts, to_ts, preset):
    """Parse from/to dates, applying preset if given. Returns (from_ts, to_ts, from_display, to_display)."""
    now = datetime.now(timezone.utc)
@@ -36,22 +68,86 @@ app = Flask(
    template_folder=os.path.join(ROOT, "templates"),
    static_folder=os.path.join(ROOT, "static"),
 )
+app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY", "dev-secret-change-in-production")
 VERSION = os.environ.get("VERSION", "dev")

+login_manager = LoginManager(app)
+login_manager.login_view = "login"
+login_manager.login_message = "Please log in to access this page."
+
+
+@login_manager.user_loader
+def load_user(user_id: str):
+    user_dict = models.get_user_by_id(int(user_id)) if user_id.isdigit() else None
+    return User(user_dict) if user_dict else None
+
+
+@login_manager.unauthorized_handler
+def unauthorized():
+    if request.is_json or request.accept_mimetypes.best == "application/json":
+        return {"error": "Authentication required"}, 401
+    return redirect(url_for("login", next=request.url))
+
+
+@app.route("/login", methods=["GET", "POST"])
+def login():
+    if current_user.is_authenticated:
+        return redirect(url_for("dashboard"))
+    if request.method == "POST":
+        username = (request.form.get("username") or "").strip()
+        password = request.form.get("password") or ""
+        user_dict = models.verify_user(username, password)
+        if user_dict:
+            login_user(User(user_dict))
+            next_param = request.form.get("next") or request.args.get("next")
+            next_url = next_param if _is_safe_redirect_url(next_param) else url_for("dashboard")
+            return redirect(next_url)
+        return render_template("login.html", error="Invalid username or password", version=VERSION)
+    return render_template("login.html", version=VERSION)
+
+
+@app.route("/logout")
+def logout():
+    logout_user()
+    return redirect(url_for("login"))
+
+
+@app.route("/users", methods=["GET", "POST"])
+@login_required
+def users():
+    if not current_user.is_admin:
+        return "Forbidden", 403
+    if request.method == "POST":
+        username = (request.form.get("username") or "").strip()
+        password = request.form.get("password") or ""
+        user_id = models.create_user(username, password, is_admin=False)
+        if user_id:
+            return redirect(url_for("users"))
+        return render_template(
+            "users.html",
+            users=models.list_users(),
+            error="Username already exists or invalid",
+            version=VERSION,
+        )
+    return render_template("users.html", users=models.list_users(), version=VERSION)
+

@app.route("/")
+@login_required
 def dashboard():
    services = models.list_services()
    return render_template("dashboard.html", services=services, version=VERSION)


@app.route("/api/services", methods=["GET"])
+@login_required
 def api_list_services():
    services = models.list_services()
    return {"services": services}


@app.route("/api/services", methods=["POST"])
+@login_required
 def api_add_service():
    data = request.get_json(silent=True)
    if data is None and request.form:
@@ -77,6 +173,7 @@ def api_add_service():


@app.route("/api/services/<int:service_id>", methods=["DELETE"])
+@login_required
 def api_delete_service(service_id):
    if models.delete_service(service_id):
        return {"deleted": service_id}
@@ -84,6 +181,7 @@ def api_delete_service(service_id):


@app.route("/api/services/<int:service_id>", methods=["PATCH"])
+@login_required
 def api_update_service(service_id):
    svc = models.get_service(service_id)
    if not svc:
@@ -116,6 +214,7 @@ def api_update_service(service_id):


@app.route("/api/services/<int:service_id>")
+@login_required
 def api_get_service(service_id):
    svc = models.get_service(service_id)
    if not svc:
@@ -125,6 +224,7 @@ def api_get_service(service_id):


@app.route("/api/services/<int:service_id>/edit")
+@login_required
 def edit_service(service_id):
    svc = models.get_service(service_id)
    if not svc:
@@ -133,6 +233,7 @@ def edit_service(service_id):


@app.route("/api/services/<int:service_id>/report")
+@login_required
 def report(service_id):
    svc = models.get_service(service_id)
    if not svc:
@@ -191,6 +292,7 @@ def _format_period_label(from_display, to_display):


@app.route("/api/services/<int:service_id>/history")
+@login_required
 def api_history(service_id):
    svc = models.get_service(service_id)
    if not svc:
@@ -207,6 +309,7 @@ def api_history(service_id):


@app.route("/api/services/<int:service_id>/stats")
+@login_required
 def api_report_stats(service_id):
    """JSON report stats with optional from/to query params for date range."""
    svc = models.get_service(service_id)
--- a/app/models.py
+++ b/app/models.py
@@ -5,6 +5,8 @@ from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 from pathlib import Path

+from werkzeug.security import check_password_hash, generate_password_hash
+
 DATA_PATH = os.environ.get("DATA_PATH", "/app/data")
 DB_PATH = Path(DATA_PATH) / "monitor.db"

@@ -55,6 +57,84 @@ def _migrate_add_rollups(conn):
        conn.execute("ALTER TABLE uptime_rollups ADD COLUMN response_count INTEGER NOT NULL DEFAULT 0")


+def _migrate_add_users(conn):
+    """Create users table for authentication."""
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS users (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            username TEXT NOT NULL UNIQUE,
+            password_hash TEXT NOT NULL,
+            is_admin INTEGER NOT NULL DEFAULT 0,
+            created_at TEXT NOT NULL
+        )
+    """)
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_users_username ON users(username)")
+
+
+def _seed_admin_if_empty(conn):
+    """Create initial admin user from env if no users exist."""
+    row = conn.execute("SELECT COUNT(*) FROM users").fetchone()
+    if row[0] > 0:
+        return
+    admin_user = os.environ.get("ADMIN_USER")
+    admin_password = os.environ.get("ADMIN_PASSWORD")
+    if not admin_user or not admin_password:
+        return
+    password_hash = generate_password_hash(admin_password)
+    conn.execute(
+        "INSERT INTO users (username, password_hash, is_admin, created_at) VALUES (?, ?, 1, ?)",
+        (admin_user, password_hash, datetime.utcnow().isoformat()),
+    )
+
+
+def create_user(username: str, password: str, is_admin: bool = False) -> int | None:
+    """Create a new user. Returns user id or None if username exists."""
+    username = username.strip()
+    if not username or not password:
+        return None
+    password_hash = generate_password_hash(password)
+    with get_db() as conn:
+        try:
+            cur = conn.execute(
+                "INSERT INTO users (username, password_hash, is_admin, created_at) VALUES (?, ?, ?, ?)",
+                (username, password_hash, 1 if is_admin else 0, datetime.utcnow().isoformat()),
+            )
+            return cur.lastrowid
+        except sqlite3.IntegrityError:
+            return None
+
+
+def get_user_by_id(user_id: int) -> dict | None:
+    """Get a user by id."""
+    with get_db() as conn:
+        row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
+        return dict(row) if row else None
+
+
+def get_user_by_username(username: str) -> dict | None:
+    """Get a user by username."""
+    with get_db() as conn:
+        row = conn.execute("SELECT * FROM users WHERE username = ?", (username.strip(),)).fetchone()
+        return dict(row) if row else None
+
+
+def verify_user(username: str, password: str) -> dict | None:
+    """Verify credentials and return user dict if valid."""
+    user = get_user_by_username(username)
+    if not user or not check_password_hash(user["password_hash"], password):
+        return None
+    return user
+
+
+def list_users():
+    """Return all users (id, username, is_admin, created_at)."""
+    with get_db() as conn:
+        rows = conn.execute(
+            "SELECT id, username, is_admin, created_at FROM users ORDER BY username"
+        ).fetchall()
+        return [dict(r) for r in rows]
+
+
@contextmanager
 def get_db():
    _ensure_data_dir()
@@ -97,6 +177,8 @@ def init_db():
        _migrate_add_status(conn)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_checks_status ON checks(status)")
        _migrate_add_rollups(conn)
+        _migrate_add_users(conn)
+        _seed_admin_if_empty(conn)


 def list_services():