127 lines
5.2 KiB
Python
127 lines
5.2 KiB
Python
"""
|
|
ThinkStorm Authentication & Authorization Module
|
|
Handles Gitea OAuth2 SSO, local session cookies, bootstrap credentials, and role-based permissions.
|
|
"""
|
|
|
|
import json
|
|
import secrets
|
|
import hashlib
|
|
import time
|
|
import urllib.request
|
|
import urllib.parse
|
|
import asyncio
|
|
from typing import Optional, Dict, Any, List
|
|
from fastapi import Request, HTTPException, status, Depends
|
|
from .config import config
|
|
from .database import get_db, hash_password, get_utc_now
|
|
from .models import User, UserRole
|
|
|
|
# Active in-memory session cache (token -> session dict)
|
|
SESSION_STORE: Dict[str, Dict[str, Any]] = {}
|
|
|
|
def generate_session_token() -> str:
|
|
return secrets.token_urlsafe(32)
|
|
|
|
def create_session(user_id: int, username: str, role: str) -> str:
|
|
token = generate_session_token()
|
|
SESSION_STORE[token] = {
|
|
"user_id": user_id,
|
|
"username": username,
|
|
"role": role,
|
|
"created_at": time.time(),
|
|
"expires_at": time.time() + (86400 * 7) # 7 days
|
|
}
|
|
return token
|
|
|
|
def destroy_session(token: str):
|
|
SESSION_STORE.pop(token, None)
|
|
|
|
def get_session(token: str) -> Optional[Dict[str, Any]]:
|
|
if not token or token not in SESSION_STORE:
|
|
return None
|
|
session = SESSION_STORE[token]
|
|
if time.time() > session["expires_at"]:
|
|
SESSION_STORE.pop(token, None)
|
|
return None
|
|
return session
|
|
|
|
def authenticate_local(username: str, password: str) -> Optional[User]:
|
|
"""Authenticates using local database credentials."""
|
|
# Check bootstrap admin first
|
|
if username == "admin" and password == config.admin_bootstrap_key:
|
|
with get_db() as conn:
|
|
row = conn.execute("SELECT id, username, password_hash, role, created_at FROM users WHERE username = 'admin'").fetchone()
|
|
if row:
|
|
return User(id=row["id"], username=row["username"], password_hash=row["password_hash"], role=UserRole.ADMIN, created_at=row["created_at"])
|
|
|
|
pwd_hash = hash_password(password)
|
|
with get_db() as conn:
|
|
row = conn.execute("SELECT id, username, password_hash, role, created_at FROM users WHERE username = ? AND password_hash = ?", (username, pwd_hash)).fetchone()
|
|
if row:
|
|
return User(id=row["id"], username=row["username"], password_hash=row["password_hash"], role=UserRole(row["role"]), created_at=row["created_at"])
|
|
return None
|
|
|
|
def get_or_create_gitea_user(gitea_user_data: Dict[str, Any]) -> User:
|
|
"""Finds or creates a ThinkStorm user based on Gitea profile."""
|
|
username = gitea_user_data.get("login") or gitea_user_data.get("username")
|
|
gitea_id = gitea_user_data.get("id")
|
|
is_gitea_admin = bool(gitea_user_data.get("is_admin", False)) or (username in config.admin_usernames)
|
|
role = UserRole.ADMIN if is_gitea_admin else UserRole.USER
|
|
now = get_utc_now()
|
|
|
|
with get_db() as conn:
|
|
row = conn.execute("SELECT id, username, password_hash, role, created_at FROM users WHERE username = ? OR gitea_id = ?", (username, gitea_id)).fetchone()
|
|
if row:
|
|
# Record the verified Gitea identity when an existing local account links
|
|
# through OAuth, and promote the role when appropriate.
|
|
resolved_role = "ADMIN" if is_gitea_admin else row["role"]
|
|
conn.execute(
|
|
"UPDATE users SET gitea_id = ?, role = ? WHERE id = ?",
|
|
(gitea_id, resolved_role, row["id"])
|
|
)
|
|
return User(
|
|
id=row["id"],
|
|
username=row["username"],
|
|
password_hash=row["password_hash"],
|
|
role=UserRole(resolved_role),
|
|
created_at=row["created_at"]
|
|
)
|
|
|
|
# Insert new user
|
|
dummy_pass = hash_password(secrets.token_hex(16))
|
|
cursor = conn.execute(
|
|
"""
|
|
INSERT INTO users (username, password_hash, role, gitea_id, created_at)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(username, dummy_pass, role.value, gitea_id, now)
|
|
)
|
|
return User(id=cursor.lastrowid, username=username, password_hash=dummy_pass, role=role, created_at=now)
|
|
|
|
async def get_current_user(request: Request) -> User:
|
|
"""FastAPI Dependency: extracts current authenticated user or returns Anonymous."""
|
|
token = request.cookies.get("thinkstorm_session")
|
|
auth_header = request.headers.get("Authorization")
|
|
if not token and auth_header and auth_header.startswith("Bearer "):
|
|
token = auth_header.split(" ", 1)[1]
|
|
|
|
session = get_session(token) if token else None
|
|
if session:
|
|
return User(
|
|
id=session["user_id"],
|
|
username=session["username"],
|
|
password_hash="",
|
|
role=UserRole(session["role"])
|
|
)
|
|
return User(id=0, username="anonymous", password_hash="", role=UserRole.ANONYMOUS)
|
|
|
|
def require_authenticated(current_user: User = Depends(get_current_user)) -> User:
|
|
if current_user.role == UserRole.ANONYMOUS:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required")
|
|
return current_user
|
|
|
|
def require_admin(current_user: User = Depends(get_current_user)) -> User:
|
|
if current_user.role != UserRole.ADMIN:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Administrator privileges required")
|
|
return current_user
|