147 lines
4.0 KiB
Python
147 lines
4.0 KiB
Python
import base64
|
|
import configparser
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
from starlette.requests import Request
|
|
|
|
from app.config import BASE_DIR
|
|
|
|
|
|
SECRET_KEY = "CapaReportSecretKey2026"
|
|
AUTH_INI_PATH = BASE_DIR / "auth.ini"
|
|
DEFAULT_USERNAME = "root"
|
|
DEFAULT_PASSWORD = "Capacity"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AuthContext:
|
|
kind: str
|
|
payload: dict
|
|
|
|
|
|
def _ensure_auth_ini() -> None:
|
|
if AUTH_INI_PATH.exists():
|
|
return
|
|
|
|
cfg = configparser.ConfigParser()
|
|
cfg["auth"] = {"username": DEFAULT_USERNAME, "password": DEFAULT_PASSWORD}
|
|
with AUTH_INI_PATH.open("w", encoding="utf-8") as file:
|
|
cfg.write(file)
|
|
|
|
|
|
def get_auth_config() -> dict[str, str]:
|
|
_ensure_auth_ini()
|
|
cfg = configparser.ConfigParser()
|
|
cfg.read(AUTH_INI_PATH, encoding="utf-8")
|
|
return {
|
|
"username": cfg.get("auth", "username", fallback=DEFAULT_USERNAME),
|
|
"password": cfg.get("auth", "password", fallback=DEFAULT_PASSWORD),
|
|
}
|
|
|
|
|
|
def save_auth_password(new_password: str) -> None:
|
|
_ensure_auth_ini()
|
|
cfg = configparser.ConfigParser()
|
|
cfg.read(AUTH_INI_PATH, encoding="utf-8")
|
|
cfg.set("auth", "password", new_password)
|
|
with AUTH_INI_PATH.open("w", encoding="utf-8") as file:
|
|
cfg.write(file)
|
|
|
|
|
|
def create_jwt_token(data: dict, expires_in: int = 86400 * 30) -> str:
|
|
header = _encode_json({"alg": "HS256", "typ": "JWT"})
|
|
payload_data = data.copy()
|
|
payload_data["exp"] = int(time.time()) + expires_in
|
|
payload = _encode_json(payload_data)
|
|
signature = _sign(header, payload)
|
|
return f"{header}.{payload}.{signature}"
|
|
|
|
|
|
def verify_jwt_token(token: str) -> Optional[dict]:
|
|
try:
|
|
header, payload, signature = token.split(".")
|
|
if not hmac.compare_digest(signature, _sign(header, payload)):
|
|
return None
|
|
|
|
payload_padded = payload + "=" * ((4 - len(payload) % 4) % 4)
|
|
data = json.loads(base64.urlsafe_b64decode(payload_padded).decode())
|
|
if data.get("exp", 0) < int(time.time()):
|
|
return None
|
|
return data
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def extract_login_token(request: Request) -> str | None:
|
|
auth_header = request.headers.get("Authorization", "").strip()
|
|
if auth_header.startswith("Bearer "):
|
|
token = auth_header.split(" ", 1)[1].strip()
|
|
if token:
|
|
return token
|
|
|
|
cookie_token = request.cookies.get("token", "").strip()
|
|
return cookie_token or None
|
|
|
|
|
|
def extract_access_token(request: Request) -> str | None:
|
|
auth_header = request.headers.get("Authorization", "").strip()
|
|
if auth_header.startswith("Bearer "):
|
|
token = auth_header.split(" ", 1)[1].strip()
|
|
if token:
|
|
return token
|
|
|
|
api_token = request.headers.get("X-API-Token", "").strip()
|
|
if api_token:
|
|
return api_token
|
|
|
|
cookie_token = request.cookies.get("token", "").strip()
|
|
return cookie_token or None
|
|
|
|
|
|
def resolve_login_context(request: Request) -> AuthContext | None:
|
|
token = extract_login_token(request)
|
|
if not token:
|
|
return None
|
|
|
|
payload = verify_jwt_token(token)
|
|
if not payload:
|
|
return None
|
|
return AuthContext(kind="jwt", payload=payload)
|
|
|
|
|
|
def resolve_access_context(request: Request) -> AuthContext | None:
|
|
token = extract_access_token(request)
|
|
if not token:
|
|
return None
|
|
|
|
payload = verify_jwt_token(token)
|
|
if payload:
|
|
return AuthContext(kind="jwt", payload=payload)
|
|
|
|
from app.services.api_tokens import verify_api_token
|
|
|
|
api_payload = verify_api_token(token)
|
|
if api_payload:
|
|
return AuthContext(kind="api_token", payload=api_payload)
|
|
|
|
return None
|
|
|
|
|
|
def _encode_json(data: dict) -> str:
|
|
raw = json.dumps(data, separators=(",", ":")).encode()
|
|
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
|
|
|
|
|
|
def _sign(header: str, payload: str) -> str:
|
|
raw = hmac.new(
|
|
SECRET_KEY.encode(),
|
|
f"{header}.{payload}".encode(),
|
|
hashlib.sha256,
|
|
).digest()
|
|
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
|