"""One-owner authentication for the SiteForge Docker Web edition. Session cookie is signed and HttpOnly. JSON mutations additionally require a session-bound CSRF header; this is intentionally a small-team / single-admin MVP, not a multi-tenant identity provider. """ import hashlib import hmac import os import secrets import time from collections import defaultdict from pathlib import Path from itsdangerous import URLSafeTimedSerializer, BadSignature, SignatureExpired from fastapi import HTTPException COOKIE = 'siteforge_session' MAX_AGE = 8 * 3600 SERIALIZER = None PASSWORD_HASH = b'' CSRF_KEY = b'' COOKIE_SECURE = False FAILURES = defaultdict(list) def configure(): global SERIALIZER, PASSWORD_HASH, CSRF_KEY, COOKIE_SECURE password_file = os.environ.get('SITEFORGE_ADMIN_PASSWORD_FILE', '').strip() if password_file: password = Path(password_file).read_text(encoding='utf-8').strip('\r\n') else: password = os.environ.get('SITEFORGE_ADMIN_PASSWORD', '') secret = os.environ.get('SITEFORGE_SESSION_SECRET', '') weak = {'password', 'admin123456', 'change-me', 'replace-me', 'changeme', '123456789', 'REPLACE_WITH_STRONG_PASSWORD'} if len(password) < 12 or password in weak: raise RuntimeError('请设置至少 12 位的 SITEFORGE_ADMIN_PASSWORD,或配置 SITEFORGE_ADMIN_PASSWORD_FILE') if len(secret) < 32 or secret in weak or secret == 'REPLACE_WITH_64_RANDOM_CHARACTERS': raise RuntimeError('请配置至少 32 位的随机 SITEFORGE_SESSION_SECRET') SERIALIZER = URLSafeTimedSerializer(secret_key=secret, salt='siteforge-web-v02') PASSWORD_HASH = hashlib.sha256(password.encode('utf-8')).digest() CSRF_KEY = hashlib.sha256(('siteforge-csrf-v02:'+secret).encode()).digest() COOKIE_SECURE = os.environ.get('SITEFORGE_COOKIE_SECURE', 'false').lower() in ('1', 'true', 'yes') FAILURES.clear() def check_password(password): return hmac.compare_digest(hashlib.sha256(password.encode('utf-8')).digest(), PASSWORD_HASH) def sign_session(): session_id = secrets.token_urlsafe(32) return SERIALIZER.dumps({'sub': 'admin', 'sid': session_id, 'v': 2}) def sign_user_session(user_id): return SERIALIZER.dumps({'sub':'user','user_id':user_id,'sid':secrets.token_urlsafe(32),'v':3}) def read_session(cookie): if not cookie or SERIALIZER is None: return None try: value = SERIALIZER.loads(cookie, max_age=MAX_AGE) if (value.get('sub'),value.get('v')) not in (('admin',2),('user',3)): return None return value except (BadSignature, SignatureExpired, ValueError, TypeError): return None def csrf(session): return hmac.new(CSRF_KEY, session['sid'].encode(), hashlib.sha256).hexdigest() def verify_csrf(session, supplied): return bool(supplied) and hmac.compare_digest(csrf(session), supplied) def check_login_rate(remote): now = time.monotonic() FAILURES[remote] = [t for t in FAILURES[remote] if now-t < 900] if len(FAILURES[remote]) >= 8: raise HTTPException(429, '登录尝试次数过多,请稍后重试') def note_login_failure(remote): FAILURES[remote].append(time.monotonic()) def note_login_success(remote): FAILURES.pop(remote, None)