feat(core): implement secure MVP core with dictionary search, auth, and CRUD

2026-09-22 13:27:11 +00:00
1 parent 4dbc53890a
commit 813d567b0e
@@ -0,0 +1,116 @@
+import hashlib
+import os
+import secrets
+import time
+from datetime import datetime, timezone, timedelta
+from typing import Optional
+from app import config
+from app.database import get_db
+
+# In-memory IP-based rate limiting for login attempts
+_login_failures: dict[str, list[float]] = {}
+
+def hash_password(password: str, salt: Optional[str] = None) -> tuple[str, str]:
+    if not salt:
+        salt = secrets.token_hex(16)
+    salt_bytes = bytes.fromhex(salt)
+    key = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt_bytes, 100_000)
+    return key.hex(), salt
+
+def verify_password(password: str, stored_hash: str, salt: str) -> bool:
+    new_hash, _ = hash_password(password, salt)
+    return secrets.compare_digest(new_hash, stored_hash)
+
+def check_login_rate_limit(ip: str) -> bool:
+    now = time.time()
+    cutoff = now - config.RATE_LIMIT_LOGIN_WINDOW_SEC
+    attempts = _login_failures.get(ip, [])
+    # Filter attempts within the time window
+    recent_attempts = [t for t in attempts if t > cutoff]
+    _login_failures[ip] = recent_attempts
+    return len(recent_attempts) < config.RATE_LIMIT_LOGIN_MAX
+
+def record_login_failure(ip: str):
+    now = time.time()
+    if ip not in _login_failures:
+        _login_failures[ip] = []
+    _login_failures[ip].append(now)
+
+def record_login_success(ip: str):
+    if ip in _login_failures:
+        _login_failures.pop(ip, None)
+
+def create_user(username: str, password: str) -> int:
+    pwd_hash, salt = hash_password(password)
+    with get_db() as conn:
+        cursor = conn.execute(
+            "INSERT INTO users (username, password_hash, salt) VALUES (?, ?, ?)",
+            (username.strip(), pwd_hash, salt)
+        )
+        return cursor.lastrowid
+
+def authenticate_user(username: str, password: str) -> Optional[dict]:
+    with get_db() as conn:
+        row = conn.execute(
+            "SELECT id, username, password_hash, salt FROM users WHERE username = ?",
+            (username.strip(),)
+        ).fetchone()
+        if not row:
+            return None
+        if verify_password(password, row["password_hash"], row["salt"]):
+            return {"id": row["id"], "username": row["username"]}
+        return None
+
+def create_session(user_id: int, user_agent: str = "", ip: str = "") -> str:
+    token = secrets.token_urlsafe(32)
+    expires_at = (datetime.now(timezone.utc) + timedelta(days=config.SESSION_EXPIRE_DAYS)).isoformat()
+    with get_db() as conn:
+        conn.execute(
+            "INSERT INTO sessions (token, user_id, expires_at, user_agent, ip) VALUES (?, ?, ?, ?, ?)",
+            (token, user_id, expires_at, user_agent, ip)
+        )
+    return token
+
+def get_user_from_session(token: str) -> Optional[dict]:
+    if not token:
+        return None
+    with get_db() as conn:
+        row = conn.execute(
+            """
+            SELECT u.id, u.username, s.expires_at
+            FROM sessions s
+            JOIN users u ON s.user_id = u.id
+            WHERE s.token = ?
+            """,
+            (token,)
+        ).fetchone()
+        if not row:
+            return None
+        
+        expires_at_str = row["expires_at"]
+        try:
+            expires_at = datetime.fromisoformat(expires_at_str)
+            if expires_at.tzinfo is None:
+                expires_at = expires_at.replace(tzinfo=timezone.utc)
+        except Exception:
+            return None
+
+        if expires_at < datetime.now(timezone.utc):
+            # Session expired
+            delete_session(token)
+            return None
+
+        return {"id": row["id"], "username": row["username"]}
+
+def delete_session(token: str):
+    if not token:
+        return
+    with get_db() as conn:
+        conn.execute("DELETE FROM sessions WHERE token = ?", (token,))
+
+def ensure_admin_user():
+    """Ensures at least one user exists; creates default admin from config if table empty."""
+    with get_db() as conn:
+        count = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
+        if count == 0 and config.ADMIN_PASSWORD:
+            create_user(config.ADMIN_USERNAME, config.ADMIN_PASSWORD)