242 lines
9.2 KiB
Python
242 lines
9.2 KiB
Python
"""Single-user, signed-cookie authentication for the PrimeThai web app."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from http.cookies import SimpleCookie
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import time
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
ENV_PATH = Path(__file__).resolve().parent.parent / ".env"
|
|
COOKIE_NAME = "primethai_session"
|
|
_ENV_KEY = re.compile(r"[A-Za-z_][A-Za-z0-9_]*$")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AuthConfig:
|
|
username: str
|
|
password: str = field(repr=False)
|
|
session_secret: bytes = field(repr=False)
|
|
session_ttl_seconds: int = 7 * 24 * 60 * 60
|
|
cookie_secure: str = "auto"
|
|
allowed_origins: frozenset[str] = field(default_factory=frozenset)
|
|
|
|
|
|
def normalize_origin(value: str) -> str | None:
|
|
"""Return a canonical HTTP(S) origin, rejecting full URLs and malformed hosts."""
|
|
value = value.strip()
|
|
if not value or "\\" in value or any(char.isspace() for char in value):
|
|
return None
|
|
try:
|
|
parsed = urlsplit(value)
|
|
port = parsed.port
|
|
except ValueError:
|
|
return None
|
|
scheme = parsed.scheme.casefold()
|
|
if (
|
|
scheme not in {"http", "https"}
|
|
or parsed.hostname is None
|
|
or parsed.username is not None
|
|
or parsed.password is not None
|
|
or parsed.path not in {"", "/"}
|
|
or parsed.query
|
|
or parsed.fragment
|
|
):
|
|
return None
|
|
host = parsed.hostname.casefold()
|
|
if ":" in host:
|
|
host = f"[{host}]"
|
|
else:
|
|
try:
|
|
host = host.encode("idna").decode("ascii").rstrip(".")
|
|
except UnicodeError:
|
|
return None
|
|
if not host:
|
|
return None
|
|
if port is not None and not ((scheme == "http" and port == 80) or (scheme == "https" and port == 443)):
|
|
host = f"{host}:{port}"
|
|
return f"{scheme}://{host}"
|
|
|
|
|
|
def _allowed_origins_from_env(raw: str) -> frozenset[str]:
|
|
allowed: set[str] = set()
|
|
for entry in raw.split(","):
|
|
entry = entry.strip()
|
|
if not entry:
|
|
continue
|
|
normalized = normalize_origin(entry)
|
|
if normalized is None:
|
|
raise RuntimeError(
|
|
"PRIMETHAI_ALLOWED_ORIGINS ต้องเป็นรายการ origin แบบ http(s)://ชื่อโฮสต์[:พอร์ต] คั่นด้วยจุลภาค"
|
|
)
|
|
allowed.add(normalized)
|
|
return frozenset(allowed)
|
|
|
|
|
|
def _dotenv_value(raw: str) -> str:
|
|
value = raw.strip()
|
|
if not value:
|
|
return ""
|
|
if value[0] in {"'", '"'}:
|
|
quote = value[0]
|
|
if quote == '"':
|
|
try:
|
|
parsed, end = json.JSONDecoder().raw_decode(value)
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError("ค่าใน .env มีเครื่องหมายคำพูดปิดไม่ครบ") from exc
|
|
trailing = value[end:].strip()
|
|
else:
|
|
end = value.rfind(quote)
|
|
if end < 1:
|
|
raise ValueError("ค่าใน .env มีเครื่องหมายคำพูดปิดไม่ครบ")
|
|
parsed = value[1:end].replace("\\'", "'").replace("\\\\", "\\")
|
|
trailing = value[end + 1:].strip()
|
|
if trailing and not trailing.startswith("#"):
|
|
raise ValueError("ค่าใน .env มีข้อความต่อท้ายหลังเครื่องหมายคำพูด")
|
|
return parsed
|
|
return re.split(r"\s+#", value, maxsplit=1)[0].rstrip()
|
|
|
|
|
|
def load_dotenv(path: str | Path = ENV_PATH) -> None:
|
|
"""Load simple KEY=VALUE entries without overriding process environment."""
|
|
env_path = Path(path)
|
|
try:
|
|
lines = env_path.read_text(encoding="utf-8").splitlines()
|
|
except FileNotFoundError:
|
|
return
|
|
for line_number, raw in enumerate(lines, start=1):
|
|
stripped = raw.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
if stripped.startswith("export "):
|
|
stripped = stripped[7:].lstrip()
|
|
key, separator, value = stripped.partition("=")
|
|
key = key.strip()
|
|
if not separator or not _ENV_KEY.fullmatch(key):
|
|
raise ValueError(f"รูปแบบ .env ไม่ถูกต้องที่บรรทัด {line_number}")
|
|
os.environ.setdefault(key, _dotenv_value(value))
|
|
|
|
|
|
def load_auth_config(path: str | Path = ENV_PATH) -> AuthConfig:
|
|
load_dotenv(path)
|
|
username = os.environ.get("PRIMETHAI_USERNAME", "").strip()
|
|
password = os.environ.get("PRIMETHAI_PASSWORD", "")
|
|
secret = os.environ.get("PRIMETHAI_SESSION_SECRET", "")
|
|
missing = [name for name, value in (
|
|
("PRIMETHAI_USERNAME", username),
|
|
("PRIMETHAI_PASSWORD", password),
|
|
("PRIMETHAI_SESSION_SECRET", secret),
|
|
) if not value]
|
|
if missing:
|
|
raise RuntimeError(
|
|
"ต้องกำหนด " + ", ".join(missing)
|
|
+ " ในไฟล์ .env ก่อนเริ่มระบบ (ดู .env.example)"
|
|
)
|
|
if len(password) < 12:
|
|
raise RuntimeError("PRIMETHAI_PASSWORD ต้องมีอย่างน้อย 12 ตัวอักษร")
|
|
if len(secret.encode("utf-8")) < 32:
|
|
raise RuntimeError("PRIMETHAI_SESSION_SECRET ต้องมีอย่างน้อย 32 ไบต์")
|
|
try:
|
|
ttl_hours = int(os.environ.get("PRIMETHAI_SESSION_TTL_HOURS", "168"))
|
|
except ValueError as exc:
|
|
raise RuntimeError("PRIMETHAI_SESSION_TTL_HOURS ต้องเป็นจำนวนเต็ม") from exc
|
|
if not 1 <= ttl_hours <= 720:
|
|
raise RuntimeError("PRIMETHAI_SESSION_TTL_HOURS ต้องอยู่ระหว่าง 1 ถึง 720")
|
|
cookie_secure = os.environ.get("PRIMETHAI_COOKIE_SECURE", "auto").strip().casefold()
|
|
if cookie_secure not in {"auto", "true", "false"}:
|
|
raise RuntimeError("PRIMETHAI_COOKIE_SECURE ต้องเป็น auto, true หรือ false")
|
|
allowed_origins = _allowed_origins_from_env(os.environ.get("PRIMETHAI_ALLOWED_ORIGINS", ""))
|
|
return AuthConfig(
|
|
username=username,
|
|
password=password,
|
|
session_secret=secret.encode("utf-8"),
|
|
session_ttl_seconds=ttl_hours * 60 * 60,
|
|
cookie_secure=cookie_secure,
|
|
allowed_origins=allowed_origins,
|
|
)
|
|
|
|
|
|
def _b64encode(data: bytes) -> str:
|
|
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
|
|
|
|
|
|
def _b64decode(value: str) -> bytes:
|
|
return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
|
|
|
|
|
|
def _credential_version(config: AuthConfig) -> str:
|
|
material = config.username.encode("utf-8") + b"\0" + config.password.encode("utf-8")
|
|
return _b64encode(hmac.new(config.session_secret, material, hashlib.sha256).digest())
|
|
|
|
|
|
def create_session(config: AuthConfig, *, now: int | None = None) -> str:
|
|
issued_at = int(time.time() if now is None else now)
|
|
payload = json.dumps(
|
|
{"sub": config.username, "cv": _credential_version(config),
|
|
"iat": issued_at, "exp": issued_at + config.session_ttl_seconds},
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
encoded = _b64encode(payload)
|
|
signature = hmac.new(config.session_secret, encoded.encode("ascii"), hashlib.sha256).digest()
|
|
return encoded + "." + _b64encode(signature)
|
|
|
|
|
|
def verify_session(config: AuthConfig, token: str | None, *, now: int | None = None) -> bool:
|
|
if not token or len(token) > 2048 or token.count(".") != 1:
|
|
return False
|
|
encoded, supplied_signature = token.split(".", 1)
|
|
expected_signature = _b64encode(
|
|
hmac.new(config.session_secret, encoded.encode("ascii", "ignore"), hashlib.sha256).digest()
|
|
)
|
|
if not hmac.compare_digest(supplied_signature, expected_signature):
|
|
return False
|
|
try:
|
|
payload = json.loads(_b64decode(encoded).decode("utf-8"))
|
|
current_time = int(time.time() if now is None else now)
|
|
return (
|
|
isinstance(payload, dict)
|
|
and hmac.compare_digest(str(payload.get("sub", "")).encode("utf-8"),
|
|
config.username.encode("utf-8"))
|
|
and hmac.compare_digest(str(payload.get("cv", "")), _credential_version(config))
|
|
and isinstance(payload.get("exp"), int)
|
|
and payload["exp"] > current_time
|
|
and isinstance(payload.get("iat"), int)
|
|
and payload["iat"] <= current_time + 60
|
|
)
|
|
except (ValueError, TypeError, UnicodeDecodeError, json.JSONDecodeError):
|
|
return False
|
|
|
|
|
|
def session_cookie(config: AuthConfig, token: str, *, secure: bool) -> str:
|
|
cookie = SimpleCookie()
|
|
cookie[COOKIE_NAME] = token
|
|
morsel = cookie[COOKIE_NAME]
|
|
morsel["path"] = "/"
|
|
morsel["max-age"] = str(config.session_ttl_seconds)
|
|
morsel["httponly"] = True
|
|
morsel["samesite"] = "Strict"
|
|
if secure:
|
|
morsel["secure"] = True
|
|
return cookie.output(header="").strip()
|
|
|
|
|
|
def expired_session_cookie(*, secure: bool) -> str:
|
|
cookie = SimpleCookie()
|
|
cookie[COOKIE_NAME] = ""
|
|
morsel = cookie[COOKIE_NAME]
|
|
morsel["path"] = "/"
|
|
morsel["max-age"] = "0"
|
|
morsel["httponly"] = True
|
|
morsel["samesite"] = "Strict"
|
|
if secure:
|
|
morsel["secure"] = True
|
|
return cookie.output(header="").strip()
|