lobbster1234/error-code-search
0
1"""2Admin authentication.3 4Primary path: verify account+password against the 【管理員登入】 sheet via the5Apps Script POST endpoint (the password never leaves Google — Apps Script only6returns ok / name / email).7 8Fallback path (no APPS_SCRIPT_URL, e.g. local dev): a single ADMIN_PASSWORD env.9 10On success we mint a *stateless* signed token carrying the admin's identity11(name + email), so later requests know who is acting without a session store.12"""13from __future__ import annotations14 15import base6416import hashlib17import hmac18import json19import os20import time21 22import httpx23 24 25def _secret() -> bytes:26 val = (27 os.getenv("ADMIN_SECRET")28 or os.getenv("APPS_SCRIPT_KEY")29 or os.getenv("ADMIN_PASSWORD")30 or "change-me"31 )32 return val.encode()33 34 35# ── Stateless signed token ──────────────────────────────────────────────────36 37def make_token(identity: dict) -> str:38 payload = json.dumps(identity, separators=(",", ":"), ensure_ascii=False).encode("utf-8")39 b = base64.urlsafe_b64encode(payload).decode()40 sig = hmac.new(_secret(), b.encode(), hashlib.sha256).hexdigest()41 return f"{b}.{sig}"42 43 44def verify_token(token: str) -> dict | None:45 if not token or "." not in token:46 return None47 b, _, sig = token.partition(".")48 expected = hmac.new(_secret(), b.encode(), hashlib.sha256).hexdigest()49 if not hmac.compare_digest(sig, expected):50 return None51 try:52 return json.loads(base64.urlsafe_b64decode(b.encode()).decode("utf-8"))53 except Exception: # noqa: BLE00154 return None55 56 57# ── Action tokens (e.g. delete-approval links) with an expiry ───────────────58 59def make_action_token(payload: dict, ttl_seconds: int = 7 * 24 * 3600) -> str:60 data = dict(payload)61 data["exp"] = int(time.time()) + ttl_seconds62 return make_token(data)63 64 65def verify_action_token(token: str) -> dict | None:66 data = verify_token(token)67 if not data:68 return None69 if int(data.get("exp", 0)) < int(time.time()):70 return None71 return data72 73 74# ── Password strength (mirrors Apps Script rules; both ends must agree) ─────75 76PASSWORD_MIN_LEN = 677COMMON_PASSWORDS = {78 "password", "passwd", "abc123", "a123456", "aa123456", "abcd1234",79 "admin", "admin123", "root", "letmein", "iloveyou", "welcome",80 "qwerty", "asdfgh", "zxcvbn", "1qaz2wsx", "q1w2e3r4", "test123",81}82 83 84def password_strength_error(pw: str) -> str | None:85 """Return a Chinese error message if the password is too weak, else None."""86 if not pw or len(pw) < PASSWORD_MIN_LEN:87 return f"密碼至少需要 {PASSWORD_MIN_LEN} 個字元"88 if pw.isdigit():89 return "密碼不可為純數字"90 if pw.lower() in COMMON_PASSWORDS:91 return "此密碼太常見,請換一個"92 return None93 94 95# ── Credential check ────────────────────────────────────────────────────────96 97async def check_login(account: str, password: str) -> dict:98 """Check credentials against the 管理員登入 sheet via Apps Script.99 100 Returns {"status": ..., "identity": ...} where status is one of:101 ok — success, identity = {name, email, account}102 needs_setup — the account exists but its password cell is empty103 (first login: the user must set a password first)104 locked — too many failed attempts; rejected without checking105 fail — wrong account/password106 107 The ADMIN_PASSWORD break-glass is only consulted when the Apps Script108 path gave no definitive answer (not configured, unreachable, or sheet109 error) — otherwise it would let an attacker keep guessing after the110 sheet-side lockout kicked in.111 """112 url = os.getenv("APPS_SCRIPT_URL", "").strip()113 114 if url:115 payload = {116 "action": "login",117 "account": account,118 "password": password,119 "key": os.getenv("APPS_SCRIPT_KEY", ""),120 }121 try:122 async with httpx.AsyncClient(timeout=20, follow_redirects=True) as client:123 resp = await client.post(url, json=payload)124 data = resp.json()125 except Exception: # noqa: BLE001 — network / parse error126 data = None127 if isinstance(data, dict) and "error" not in data:128 if data.get("ok"):129 return {"status": "ok", "identity": {130 "name": (data.get("name") or account).strip(),131 "email": (data.get("email") or "").strip(),132 "account": account,133 }}134 if data.get("locked"):135 return {"status": "locked"}136 if data.get("needsSetup"):137 return {"status": "needs_setup"}138 return {"status": "fail"} # definitive rejection — no break-glass139 140 # Break-glass fallback: single env password, only when the sheet path141 # is unavailable so there is still a way in.142 real = os.getenv("ADMIN_PASSWORD", "")143 if real and hmac.compare_digest(password, real):144 return {"status": "ok", "identity": {145 "name": account or "admin",146 "email": os.getenv("ADMIN_EMAIL", ""),147 "account": account or "admin",148 }}149 return {"status": "fail"}150 151 152async def setup_password(account: str, new_password: str) -> dict:153 """First-time password setup, forwarded to Apps Script.154 155 The real guard lives in Apps Script: it only writes when the account's156 password cell is empty, so an existing password can never be overwritten157 through this entry point. Returns the parsed Apps Script response, e.g.158 {ok: true, name, email} or {ok: false, error: "already_set" | ...}.159 """160 res = await apps_script_call(161 "setupPassword", {"account": account, "newPassword": new_password}162 )163 if not isinstance(res, dict):164 return {"ok": False, "error": "unreachable"}165 return res166 167 168def auth_configured() -> bool:169 return bool(os.getenv("APPS_SCRIPT_URL") or os.getenv("ADMIN_PASSWORD"))170 171 172async def apps_script_call(action: str, payload: dict | None = None) -> dict | None:173 """Generic POST to the Apps Script web app. Returns parsed JSON or None."""174 url = os.getenv("APPS_SCRIPT_URL", "").strip()175 if not url:176 return None177 body = {"action": action, "key": os.getenv("APPS_SCRIPT_KEY", "")}178 if payload:179 body.update(payload)180 try:181 async with httpx.AsyncClient(timeout=20, follow_redirects=True) as client:182 resp = await client.post(url, json=body)183 return resp.json()184 except Exception: # noqa: BLE001185 return None186 187 188async def _fetch_people(action: str, key: str) -> list[dict]:189 url = os.getenv("APPS_SCRIPT_URL", "").strip()190 if not url:191 return []192 payload = {"action": action, "key": os.getenv("APPS_SCRIPT_KEY", "")}193 try:194 async with httpx.AsyncClient(timeout=20, follow_redirects=True) as client:195 resp = await client.post(url, json=payload)196 data = resp.json()197 if isinstance(data, dict) and data.get("ok"):198 return [p for p in data.get(key, []) if p.get("email")]199 except Exception: # noqa: BLE001200 pass201 return []202 203 204async def fetch_approvers() -> list[dict]:205 """Read the 審核員 sheet via Apps Script → [{name, email}]."""206 return await _fetch_people("getApprovers", "approvers")207 208 209async def fetch_members() -> list[dict]:210 """Read the 管理員登入 sheet via Apps Script → MT members [{name, email}] (no passwords)."""211 return await _fetch_people("getMembers", "members")212 