浅念
3c57c167f2
导入 SiteForge Factory V5 基线源码： - FastAPI + SQLite + Docker Compose（Web + Worker）架构 - 四套商业主题 × 14 页面模板（1.1.0，含历史版本快照） - 四种建站模式骨架 + 迁移审计 + 发布门禁 - 52 项 pytest 测试 - 补全计划主文档 docs/superpowers/plans/ Co-Authored-By: Claude Code <noreply@anthropic.com>
105 lines
3.9 KiB
Python
105 lines
3.9 KiB
Python
"""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''
|
|
PASSWORD_SALT = b''
|
|
CSRF_KEY = b''
|
|
COOKIE_SECURE = False
|
|
FAILURES = defaultdict(list)
|
|
|
|
|
|
def configure():
|
|
global SERIALIZER, PASSWORD_HASH, PASSWORD_SALT, 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')
|
|
# Keep the deployment secret out of memory after startup and make each
|
|
# verification deliberately expensive, consistent with team accounts.
|
|
PASSWORD_SALT = secrets.token_bytes(16)
|
|
PASSWORD_HASH = hashlib.scrypt(password.encode('utf-8'),salt=PASSWORD_SALT,n=2**14,r=8,p=1,dklen=32)
|
|
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):
|
|
try:
|
|
candidate=hashlib.scrypt(password.encode('utf-8'),salt=PASSWORD_SALT,n=2**14,r=8,p=1,dklen=32)
|
|
return hmac.compare_digest(candidate,PASSWORD_HASH)
|
|
except (ValueError,TypeError,MemoryError):
|
|
return False
|
|
|
|
|
|
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()
|
|
# Drop expired buckets so spoofed source addresses cannot grow memory
|
|
# without bound on a long-running process.
|
|
for key in list(FAILURES):
|
|
recent=[t for t in FAILURES[key] if now-t < 900]
|
|
if recent:FAILURES[key]=recent
|
|
else:FAILURES.pop(key,None)
|
|
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)
|