"""Command Sovereignty auth server — Flask + SQLite + bcrypt."""
import os
import secrets
import sqlite3
from datetime import datetime, timezone
import bcrypt
from flask import Flask, Response, redirect, request, send_from_directory, session, jsonify
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
PUBLIC_DIR = os.path.join(BASE_DIR, "public")
DB_PATH = os.path.join(BASE_DIR, "auth.db")
SECRET_KEY = secrets.token_hex(32)
app = Flask(__name__, static_folder=PUBLIC_DIR, static_url_path="")
app.secret_key = SECRET_KEY
def get_db():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
return conn
def init_db():
conn = get_db()
conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
display_name TEXT,
created_at TEXT DEFAULT (datetime('now'))
)
""")
conn.commit()
conn.close()
@app.route("/api/auth/signup", methods=["POST"])
def signup():
data = request.get_json()
email = (data.get("email") or "").strip().lower()
password = data.get("password") or ""
name = data.get("display_name") or email
if not email or not password:
return jsonify(error="Email and password required"), 400
if len(password) < 8:
return jsonify(error="Password must be at least 8 characters"), 400
conn = get_db()
try:
existing = conn.execute("SELECT id FROM users WHERE email = ?", (email,)).fetchone()
if existing:
return jsonify(error="Account already exists"), 409
password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
conn.execute(
"INSERT INTO users (email, password_hash, display_name) VALUES (?, ?, ?)",
(email, password_hash.decode(), name),
)
conn.commit()
# Auto-log in on signup
user = conn.execute("SELECT id, email, display_name FROM users WHERE email = ?", (email,)).fetchone()
session.clear()
session["user_id"] = user["id"]
session["email"] = user["email"]
session["name"] = user["display_name"]
session["logged_in"] = True
return jsonify(user={"email": user["email"], "name": user["display_name"]}), 201
finally:
conn.close()
@app.route("/api/auth/login", methods=["POST"])
def login():
data = request.get_json()
email = (data.get("email") or "").strip().lower()
password = data.get("password") or ""
if not email or not password:
return jsonify(error="Email and password required"), 400
conn = get_db()
try:
user = conn.execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone()
if not user or not bcrypt.checkpw(password.encode(), user["password_hash"].encode()):
return jsonify(error="Invalid email or password"), 401
session.clear()
session["user_id"] = user["id"]
session["email"] = user["email"]
session["name"] = user["display_name"]
session["logged_in"] = True
return jsonify(user={"email": user["email"], "name": user["display_name"]})
finally:
conn.close()
@app.route("/api/auth/logout", methods=["POST"])
def logout():
session.clear()
return jsonify(ok=True)
@app.route("/api/auth/me")
def me():
if not session.get("logged_in"):
return jsonify(user=None), 401
return jsonify(user={
"email": session.get("email"),
"name": session.get("name"),
})
@app.route("/api/auth/me", methods=["PUT"])
def update_profile():
if not session.get("logged_in"):
return jsonify(error="Not authenticated"), 401
data = request.get_json()
display_name = data.get("display_name")
if display_name is None:
return jsonify(error="display_name required"), 400
display_name = display_name.strip()
if not display_name:
return jsonify(error="display_name cannot be empty"), 400
conn = get_db()
try:
conn.execute(
"UPDATE users SET display_name = ? WHERE id = ?",
(display_name, session["user_id"]),
)
conn.commit()
session["name"] = display_name
return jsonify(user={
"email": session.get("email"),
"name": session.get("name"),
})
finally:
conn.close()
@app.route("/api/auth/create-admin", methods=["POST"])
def create_admin():
"""Create a demo admin user — disable in production."""
data = request.get_json()
email = data.get("email", "demo@commandsovereignty.com")
password = data.get("password", "demo123")
conn = get_db()
try:
existing = conn.execute("SELECT id FROM users WHERE email = ?", (email,)).fetchone()
if existing:
return jsonify(message="User already exists"), 200
password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
conn.execute(
"INSERT INTO users (email, password_hash, display_name) VALUES (?, ?, ?)",
(email, password_hash.decode(), email),
)
conn.commit()
return jsonify(message="Admin user created"), 201
finally:
conn.close()
# Serve static files with correct MIME types
@app.route("/")
def index():
return send_from_directory(PUBLIC_DIR, "index.html")
@app.route("/dashboard")
def dashboard():
if not session.get("logged_in"):
return redirect("/")
return send_from_directory(PUBLIC_DIR, "dashboard.html")
# Catch-all for SPA routes
@app.route("/<path:path>")
def serve_file(path):
filepath = os.path.join(PUBLIC_DIR, path)
if os.path.isfile(filepath):
return send_from_directory(PUBLIC_DIR, path)
# Try adding .html extension
html_path = path + ".html" if not path.endswith(".html") else path
html_filepath = os.path.join(PUBLIC_DIR, html_path)
if os.path.isfile(html_filepath):
return send_from_directory(PUBLIC_DIR, html_path)
# Fallback to index.html for SPA
if not path.endswith((".js", ".css", ".png", ".svg", ".woff2")):
return send_from_directory(PUBLIC_DIR, "index.html")
return "Not found", 404
if __name__ == "__main__":
init_db()
print("Auth DB initialized at:", DB_PATH)
app.run(host="0.0.0.0", port=5003, debug=False)