1471 lines
62 KiB
Python
1471 lines
62 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import secrets
|
||
import sqlite3
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from typing import Any, Optional
|
||
from urllib.parse import quote
|
||
|
||
from gateway.config import Platform
|
||
from gateway.pairing import PairingStore
|
||
from gateway.session import SessionSource
|
||
from hermes_constants import get_hermes_home
|
||
from hermes_cli.dashboard_auth.prefix import resolve_public_url
|
||
from plugins.platforms.email.adapter import _standalone_send as _email_standalone_send
|
||
|
||
_ALLOWED_DOMAINS = ("bremen.com.tw", "journeys.com.tw")
|
||
# Verification originally started as a LINE / Telegram DM gate, but the same
|
||
# canonical principal must also resolve for verified email senders and the
|
||
# OpenAI-compatible API server used by OpenWebUI-style frontends.
|
||
_SUPPORTED_PLATFORM_VALUES = {"line", "telegram", "email", "api_server"}
|
||
_EMAIL_RE = re.compile(r"(?i)\b([A-Z0-9._%+\-]+@[A-Z0-9.\-]+\.[A-Z]{2,})\b")
|
||
|
||
_STATE_NEW = "new"
|
||
_STATE_PENDING = "pending_verification"
|
||
_STATE_VERIFIED = "verified"
|
||
_STATE_COOLDOWN = "cooldown_blacklist"
|
||
_STATE_PERMANENT = "permanent_blocked"
|
||
_STATE_ADMIN_BLOCKED = "admin_blocked"
|
||
_ROLE_VIEWER = "viewer"
|
||
_ROLE_OPERATOR = "operator"
|
||
_ROLE_ADMIN = "admin"
|
||
_VALID_ROLES = (_ROLE_VIEWER, _ROLE_OPERATOR, _ROLE_ADMIN)
|
||
_SEND_COOLDOWN_SECONDS = 60
|
||
_EMAIL_WINDOW_MINUTES = 10
|
||
_EMAIL_WINDOW_MAX_SENDS = 3
|
||
|
||
|
||
@dataclass
|
||
class VerificationDecision:
|
||
action: str
|
||
message: str = ""
|
||
email: Optional[str] = None
|
||
token: Optional[str] = None
|
||
verification_url: Optional[str] = None
|
||
state: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class VerificationResult:
|
||
success: bool
|
||
message: str
|
||
state: Optional[str] = None
|
||
principal_id: Optional[str] = None
|
||
email: Optional[str] = None
|
||
|
||
|
||
def _utcnow() -> datetime:
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
def _iso(dt: datetime | None) -> str | None:
|
||
if dt is None:
|
||
return None
|
||
return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat()
|
||
|
||
|
||
def _parse_iso(value: str | None) -> datetime | None:
|
||
if not value:
|
||
return None
|
||
try:
|
||
dt = datetime.fromisoformat(value)
|
||
except ValueError:
|
||
return None
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt.astimezone(timezone.utc)
|
||
|
||
|
||
def _format_local(dt: datetime | None) -> str:
|
||
if dt is None:
|
||
return "未知時間"
|
||
return dt.astimezone().strftime("%Y-%m-%d %H:%M")
|
||
|
||
|
||
def _local_day_key(dt: datetime | None = None) -> str:
|
||
current = dt.astimezone() if dt is not None else datetime.now().astimezone()
|
||
return current.strftime("%Y-%m-%d")
|
||
|
||
|
||
def _normalize_email(email: str) -> str:
|
||
return (email or "").strip().lower()
|
||
|
||
|
||
def _source_identity_key(source: SessionSource | None) -> str:
|
||
if source is None:
|
||
return ""
|
||
platform_value = getattr(getattr(source, "platform", None), "value", getattr(source, "platform", None))
|
||
if platform_value == "email":
|
||
return _normalize_email(str(getattr(source, "user_id", "") or getattr(source, "chat_id", "") or ""))
|
||
return str(getattr(source, "user_id", "") or "").strip()
|
||
|
||
|
||
def _extract_email(text: str | None) -> str | None:
|
||
if not text:
|
||
return None
|
||
match = _EMAIL_RE.search(text.strip())
|
||
if not match:
|
||
return None
|
||
return _normalize_email(match.group(1))
|
||
|
||
|
||
def _email_domain(email: str) -> str:
|
||
return _normalize_email(email).split("@", 1)[-1]
|
||
|
||
|
||
def _is_allowed_domain(email: str) -> bool:
|
||
return _email_domain(email) in _ALLOWED_DOMAINS
|
||
|
||
|
||
def _sha256(text: str) -> str:
|
||
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def default_db_path() -> Path:
|
||
return get_hermes_home() / "gateway_users.db"
|
||
|
||
|
||
def resolve_verification_public_base_url() -> str:
|
||
"""Resolve the public dashboard base URL used in verification emails.
|
||
|
||
Only the dashboard-specific public URL is safe here. Falling back to
|
||
messaging-platform webhook hosts can produce broken links (for example a
|
||
LINE ngrok tunnel that does not serve the dashboard route), which users
|
||
see as 404s even though the email itself was delivered.
|
||
"""
|
||
dashboard_public = resolve_public_url()
|
||
if dashboard_public:
|
||
return dashboard_public.rstrip("/")
|
||
|
||
env_public = (os.getenv("HERMES_PUBLIC_BASE_URL", "") or "").strip().rstrip("/")
|
||
if env_public:
|
||
return env_public
|
||
|
||
return ""
|
||
|
||
|
||
def build_verification_url(token: str, *, public_base_url: Optional[str] = None) -> str:
|
||
base = (public_base_url or resolve_verification_public_base_url() or "").rstrip("/")
|
||
if not base:
|
||
return ""
|
||
return f"{base}/api/verification/verify?token={quote(token)}"
|
||
|
||
|
||
async def send_verification_email(
|
||
to_email: str,
|
||
*,
|
||
token: str,
|
||
verification_url: str,
|
||
platform: str,
|
||
external_user_id: str,
|
||
pconfig: Any = None,
|
||
) -> dict[str, Any]:
|
||
body_lines = [
|
||
"Hermes 驗證信",
|
||
"",
|
||
"請在 24 小時內完成驗證。",
|
||
f"平台：{platform}",
|
||
f"帳號識別：{external_user_id}",
|
||
"",
|
||
]
|
||
if verification_url:
|
||
body_lines.extend([
|
||
"點擊以下連結完成驗證：",
|
||
verification_url,
|
||
"",
|
||
])
|
||
body_lines.extend([
|
||
f"驗證代碼：{token}",
|
||
"",
|
||
"如果這不是你本人操作，請直接忽略此信。",
|
||
])
|
||
result = await _email_standalone_send(
|
||
pconfig or SimpleNamespace(extra={}),
|
||
to_email,
|
||
"\n".join(body_lines),
|
||
)
|
||
if not isinstance(result, dict):
|
||
raise RuntimeError("Email send failed: invalid response")
|
||
if not result.get("success"):
|
||
error = str(result.get("error") or "Email send failed")
|
||
raise RuntimeError(error)
|
||
return result
|
||
|
||
|
||
class _VerificationStateMachine:
|
||
"""Owns verification *decision* logic: token issuance, rate limits, and
|
||
identity_enforcement state transitions (new -> pending -> verified /
|
||
cooldown / permanent).
|
||
|
||
Deliberately has no knowledge of admin actions (block/unblock/force-bind),
|
||
dashboard roles, quotas, model policies, or usage accounting — those stay
|
||
on GatewayUserStore. This is a Phase 1 extraction (see
|
||
hermes-architecture-design-docs proposal): the public methods on
|
||
GatewayUserStore keep their exact signatures and just delegate here, so no
|
||
call site (gateway/run.py, web_server.py, adapters, tests) needs to change.
|
||
Reuses the owning store's `_connect()` / `_get_identity_row()` since those
|
||
are shared low-level DB helpers, not verification-specific decisions.
|
||
"""
|
||
|
||
def __init__(self, store: "GatewayUserStore"):
|
||
self._store = store
|
||
|
||
@staticmethod
|
||
def supports_source(source: SessionSource | None) -> bool:
|
||
return bool(
|
||
source
|
||
and source.chat_type == "dm"
|
||
and getattr(source.platform, "value", source.platform) in _SUPPORTED_PLATFORM_VALUES
|
||
and source.user_id
|
||
)
|
||
|
||
def _rate_limit_message(self, *, seconds: int | None = None, window_minutes: int | None = None, max_sends: int | None = None) -> str:
|
||
if seconds is not None:
|
||
return f"驗證信剛寄出，請稍候 {seconds} 秒後再重試。"
|
||
return f"此 Email 在 {window_minutes} 分鐘內最多只能寄送 {max_sends} 封驗證信，請稍後再試。"
|
||
|
||
def _check_send_rate_limits(
|
||
self,
|
||
conn: sqlite3.Connection,
|
||
*,
|
||
platform: str,
|
||
external_user_id: str,
|
||
email: str,
|
||
now: datetime,
|
||
) -> str | None:
|
||
identity_recent = conn.execute(
|
||
"""
|
||
SELECT last_sent_at FROM verification_requests
|
||
WHERE platform = ? AND external_user_id = ?
|
||
ORDER BY id DESC LIMIT 1
|
||
""",
|
||
(platform, external_user_id),
|
||
).fetchone()
|
||
if identity_recent is not None:
|
||
last_sent_at = _parse_iso(identity_recent["last_sent_at"])
|
||
if last_sent_at and (now - last_sent_at).total_seconds() < _SEND_COOLDOWN_SECONDS:
|
||
return self._rate_limit_message(seconds=_SEND_COOLDOWN_SECONDS)
|
||
|
||
window_start = _iso(now - timedelta(minutes=_EMAIL_WINDOW_MINUTES))
|
||
email_count = conn.execute(
|
||
"""
|
||
SELECT COUNT(1) AS c FROM verification_requests
|
||
WHERE email = ? AND last_sent_at >= ?
|
||
""",
|
||
(email, window_start),
|
||
).fetchone()
|
||
if email_count and int(email_count["c"] or 0) >= _EMAIL_WINDOW_MAX_SENDS:
|
||
return self._rate_limit_message(
|
||
window_minutes=_EMAIL_WINDOW_MINUTES,
|
||
max_sends=_EMAIL_WINDOW_MAX_SENDS,
|
||
)
|
||
return None
|
||
|
||
def _ensure_identity_row(self, conn: sqlite3.Connection, source: SessionSource) -> sqlite3.Row:
|
||
platform = source.platform.value
|
||
user_id = str(source.user_id)
|
||
row = self._store._get_identity_row(conn, platform, user_id)
|
||
now = _iso(_utcnow())
|
||
if row is None:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO identity_enforcement(
|
||
platform, external_user_id, external_user_name, state, created_at, updated_at
|
||
) VALUES (?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(platform, user_id, source.user_name, _STATE_NEW, now, now),
|
||
)
|
||
row = self._store._get_identity_row(conn, platform, user_id)
|
||
elif source.user_name and source.user_name != row["external_user_name"]:
|
||
conn.execute(
|
||
"UPDATE identity_enforcement SET external_user_name = ?, updated_at = ? WHERE id = ?",
|
||
(source.user_name, now, row["id"]),
|
||
)
|
||
row = self._store._get_identity_row(conn, platform, user_id)
|
||
if row is None:
|
||
raise RuntimeError("failed to create identity_enforcement row")
|
||
return row
|
||
|
||
def _reconcile_identity_state(self, conn: sqlite3.Connection, source: SessionSource) -> sqlite3.Row:
|
||
row = self._ensure_identity_row(conn, source)
|
||
state = row["state"]
|
||
now = _utcnow()
|
||
blocked_until = _parse_iso(row["blocked_until"])
|
||
updated = False
|
||
|
||
if state == _STATE_PENDING and row["last_request_id"]:
|
||
req = conn.execute(
|
||
"SELECT * FROM verification_requests WHERE id = ?",
|
||
(row["last_request_id"],),
|
||
).fetchone()
|
||
expires_at = _parse_iso(req["expires_at"] if req else None)
|
||
if req and req["status"] == _STATE_PENDING and expires_at and expires_at <= now:
|
||
conn.execute(
|
||
"UPDATE verification_requests SET status = 'expired' WHERE id = ?",
|
||
(req["id"],),
|
||
)
|
||
failure_cycles = int(row["failure_cycles"] or 0) + 1
|
||
next_state = _STATE_PERMANENT if failure_cycles >= 3 else _STATE_COOLDOWN
|
||
next_blocked = None if next_state == _STATE_PERMANENT else _iso(now + timedelta(hours=72))
|
||
conn.execute(
|
||
"""
|
||
UPDATE identity_enforcement
|
||
SET state = ?, failure_cycles = ?, blocked_until = ?, pending_email = NULL, updated_at = ?
|
||
WHERE id = ?
|
||
""",
|
||
(next_state, failure_cycles, next_blocked, _iso(now), row["id"]),
|
||
)
|
||
updated = True
|
||
|
||
elif state == _STATE_COOLDOWN and blocked_until and blocked_until <= now:
|
||
conn.execute(
|
||
"""
|
||
UPDATE identity_enforcement
|
||
SET state = ?, blocked_until = NULL, pending_email = NULL, updated_at = ?
|
||
WHERE id = ?
|
||
""",
|
||
(_STATE_NEW, _iso(now), row["id"]),
|
||
)
|
||
updated = True
|
||
|
||
if updated:
|
||
row = self._store._get_identity_row(conn, source.platform.value, str(source.user_id))
|
||
if row is None:
|
||
raise RuntimeError("identity_enforcement row disappeared during reconciliation")
|
||
return row
|
||
|
||
def is_verified_source(self, source: SessionSource) -> bool:
|
||
if not self.supports_source(source):
|
||
return False
|
||
with self._store._connect() as conn:
|
||
row = self._store._get_identity_row(conn, source.platform.value, str(source.user_id))
|
||
if row is None:
|
||
return False
|
||
if row["state"] != _STATE_VERIFIED:
|
||
return False
|
||
identity = conn.execute(
|
||
"SELECT 1 FROM external_identities WHERE platform = ? AND external_user_id = ?",
|
||
(source.platform.value, str(source.user_id)),
|
||
).fetchone()
|
||
return identity is not None
|
||
|
||
def get_verified_email_for_source(self, source: SessionSource) -> str | None:
|
||
"""Return the bound verified email for a verified source, if any."""
|
||
if not self.supports_source(source):
|
||
return None
|
||
with self._store._connect() as conn:
|
||
row = conn.execute(
|
||
"""
|
||
SELECT e.verified_email
|
||
FROM external_identities e
|
||
JOIN identity_enforcement i
|
||
ON i.platform = e.platform
|
||
AND i.external_user_id = e.external_user_id
|
||
WHERE e.platform = ?
|
||
AND e.external_user_id = ?
|
||
AND i.state = ?
|
||
""",
|
||
(source.platform.value, str(source.user_id), _STATE_VERIFIED),
|
||
).fetchone()
|
||
if row is None:
|
||
return None
|
||
email = str(row["verified_email"] or "").strip()
|
||
return email or None
|
||
|
||
def process_inbound_message(
|
||
self,
|
||
source: SessionSource,
|
||
text: str | None,
|
||
*,
|
||
public_base_url: Optional[str] = None,
|
||
) -> VerificationDecision:
|
||
if not self.supports_source(source):
|
||
return VerificationDecision(action="allow", state=None)
|
||
|
||
with self._store._connect() as conn:
|
||
row = self._reconcile_identity_state(conn, source)
|
||
state = row["state"]
|
||
if state == _STATE_VERIFIED:
|
||
return VerificationDecision(action="allow", state=state)
|
||
if state == _STATE_ADMIN_BLOCKED:
|
||
return VerificationDecision(
|
||
action="handled",
|
||
state=state,
|
||
message="此帳號目前已被管理員停用，請聯絡管理員。",
|
||
)
|
||
if state == _STATE_PERMANENT:
|
||
return VerificationDecision(
|
||
action="handled",
|
||
state=state,
|
||
message="此帳號已因連續三次未完成驗證而永久停用，請聯絡管理員。",
|
||
)
|
||
if state == _STATE_COOLDOWN:
|
||
blocked_until = _format_local(_parse_iso(row["blocked_until"]))
|
||
return VerificationDecision(
|
||
action="handled",
|
||
state=state,
|
||
message=(
|
||
"你目前因逾時未完成驗證而暫時停用，"
|
||
f"請於 {blocked_until} 後再重新申請驗證。"
|
||
),
|
||
)
|
||
|
||
email = _extract_email(text)
|
||
if state == _STATE_PENDING:
|
||
return VerificationDecision(
|
||
action="handled",
|
||
state=state,
|
||
message=(
|
||
"你尚未完成驗證，請於 24 小時內收信並點擊驗證連結。"
|
||
"若未完成，系統將暫停受理 72 小時。"
|
||
),
|
||
)
|
||
|
||
if not email:
|
||
return VerificationDecision(
|
||
action="handled",
|
||
state=state,
|
||
message=(
|
||
"歡迎使用，請先直接回覆你的公司 Email 以完成驗證。"
|
||
"僅接受 @bremen.com.tw 或 @journeys.com.tw；"
|
||
"例如：sunny.lin@bremen.com.tw。"
|
||
),
|
||
)
|
||
|
||
if not _is_allowed_domain(email):
|
||
return VerificationDecision(
|
||
action="handled",
|
||
state=state,
|
||
message=(
|
||
"目前只接受公司 Email 驗證。"
|
||
"請直接回覆 @bremen.com.tw 或 @journeys.com.tw 的公司信箱；"
|
||
"例如：sunny.lin@bremen.com.tw。"
|
||
),
|
||
)
|
||
|
||
platform = source.platform.value
|
||
existing = conn.execute(
|
||
"""
|
||
SELECT external_user_id FROM external_identities
|
||
WHERE platform = ? AND verified_email = ?
|
||
""",
|
||
(platform, email),
|
||
).fetchone()
|
||
if existing and existing["external_user_id"] != str(source.user_id):
|
||
return VerificationDecision(
|
||
action="handled",
|
||
state=state,
|
||
message="已經綁定。",
|
||
)
|
||
|
||
token = secrets.token_urlsafe(24)
|
||
verification_url = build_verification_url(token, public_base_url=public_base_url)
|
||
now = _utcnow()
|
||
rate_limited = self._check_send_rate_limits(
|
||
conn,
|
||
platform=platform,
|
||
external_user_id=str(source.user_id),
|
||
email=email,
|
||
now=now,
|
||
)
|
||
if rate_limited:
|
||
return VerificationDecision(
|
||
action="handled",
|
||
state=state,
|
||
email=email,
|
||
message=rate_limited,
|
||
)
|
||
expires_at = now + timedelta(hours=24)
|
||
conn.execute(
|
||
"UPDATE verification_requests SET status = 'cancelled' WHERE platform = ? AND external_user_id = ? AND status = ?",
|
||
(platform, str(source.user_id), _STATE_PENDING),
|
||
)
|
||
cur = conn.execute(
|
||
"""
|
||
INSERT INTO verification_requests(
|
||
platform, external_user_id, email, token_hash, status, expires_at, verified_at, created_at, last_sent_at
|
||
) VALUES (?, ?, ?, ?, ?, ?, NULL, ?, ?)
|
||
""",
|
||
(
|
||
platform,
|
||
str(source.user_id),
|
||
email,
|
||
_sha256(token),
|
||
_STATE_PENDING,
|
||
_iso(expires_at),
|
||
_iso(now),
|
||
_iso(now),
|
||
),
|
||
)
|
||
request_id = cur.lastrowid
|
||
conn.execute(
|
||
"""
|
||
UPDATE identity_enforcement
|
||
SET state = ?, pending_email = ?, last_request_id = ?, updated_at = ?
|
||
WHERE id = ?
|
||
""",
|
||
(_STATE_PENDING, email, request_id, _iso(now), row["id"]),
|
||
)
|
||
return VerificationDecision(
|
||
action="send_verification_email",
|
||
state=_STATE_PENDING,
|
||
email=email,
|
||
token=token,
|
||
verification_url=verification_url,
|
||
message="已寄出驗證信，請於 24 小時內收信並點擊連結完成驗證。",
|
||
)
|
||
|
||
def verify_token(self, token: str) -> VerificationResult:
|
||
token_hash = _sha256((token or "").strip())
|
||
with self._store._connect() as conn:
|
||
req = conn.execute(
|
||
"SELECT * FROM verification_requests WHERE token_hash = ?",
|
||
(token_hash,),
|
||
).fetchone()
|
||
if req is None:
|
||
return VerificationResult(False, "驗證連結無效或已失效。")
|
||
if req["status"] != _STATE_PENDING:
|
||
return VerificationResult(False, "驗證連結已使用或已失效。")
|
||
expires_at = _parse_iso(req["expires_at"])
|
||
now = _utcnow()
|
||
if expires_at and expires_at <= now:
|
||
source = SessionSource(
|
||
platform=Platform(req["platform"]),
|
||
chat_id=req["external_user_id"],
|
||
user_id=req["external_user_id"],
|
||
chat_type="dm",
|
||
)
|
||
self._reconcile_identity_state(conn, source)
|
||
return VerificationResult(False, "驗證連結已逾期，請重新申請。")
|
||
|
||
existing = conn.execute(
|
||
"SELECT external_user_id FROM external_identities WHERE platform = ? AND verified_email = ?",
|
||
(req["platform"], req["email"]),
|
||
).fetchone()
|
||
if existing and existing["external_user_id"] != req["external_user_id"]:
|
||
return VerificationResult(False, "此 Email 已經綁定其他帳號。")
|
||
|
||
principal = conn.execute(
|
||
"SELECT * FROM principals WHERE email = ?",
|
||
(req["email"],),
|
||
).fetchone()
|
||
now_iso = _iso(now)
|
||
principal_id = principal["principal_id"] if principal else secrets.token_hex(12)
|
||
if principal is None:
|
||
conn.execute(
|
||
"INSERT INTO principals(principal_id, email, status, created_at, updated_at) VALUES (?, ?, 'verified', ?, ?)",
|
||
(principal_id, req["email"], now_iso, now_iso),
|
||
)
|
||
else:
|
||
conn.execute(
|
||
"UPDATE principals SET status = 'verified', updated_at = ? WHERE principal_id = ?",
|
||
(now_iso, principal_id),
|
||
)
|
||
|
||
conn.execute(
|
||
"DELETE FROM external_identities WHERE platform = ? AND external_user_id = ?",
|
||
(req["platform"], req["external_user_id"]),
|
||
)
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO external_identities(
|
||
principal_id, platform, external_user_id, user_name, verified_email, verified_at
|
||
) VALUES (?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(principal_id, req["platform"], req["external_user_id"], None, req["email"], now_iso),
|
||
)
|
||
conn.execute(
|
||
"""
|
||
UPDATE identity_enforcement
|
||
SET state = ?, principal_id = ?, verified_email = ?, pending_email = NULL,
|
||
blocked_until = NULL, last_request_id = ?, updated_at = ?
|
||
WHERE platform = ? AND external_user_id = ?
|
||
""",
|
||
(
|
||
_STATE_VERIFIED,
|
||
principal_id,
|
||
req["email"],
|
||
req["id"],
|
||
now_iso,
|
||
req["platform"],
|
||
req["external_user_id"],
|
||
),
|
||
)
|
||
conn.execute(
|
||
"UPDATE verification_requests SET status = ?, verified_at = ? WHERE id = ?",
|
||
(_STATE_VERIFIED, now_iso, req["id"]),
|
||
)
|
||
return VerificationResult(True, "驗證成功，現在可以開始使用 Hermes。", state=_STATE_VERIFIED, principal_id=principal_id, email=req["email"])
|
||
|
||
def resend_verification(self, platform: str, external_user_id: str, *, public_base_url: Optional[str] = None) -> dict[str, Any]:
|
||
with self._store._connect() as conn:
|
||
row = self._store._get_identity_row(conn, platform, external_user_id)
|
||
if row is None or row["state"] != _STATE_PENDING or not row["pending_email"]:
|
||
raise KeyError("pending_verification_not_found")
|
||
pending_email = str(row["pending_email"])
|
||
conn.execute(
|
||
"UPDATE identity_enforcement SET state = ?, pending_email = NULL, updated_at = ? WHERE id = ?",
|
||
(_STATE_NEW, _iso(_utcnow()), row["id"]),
|
||
)
|
||
source = SessionSource(platform=Platform(platform), chat_id=external_user_id, user_id=external_user_id, user_name=row["external_user_name"], chat_type="dm")
|
||
conn.commit()
|
||
decision = self.process_inbound_message(source, pending_email, public_base_url=public_base_url)
|
||
if decision.action == "send_verification_email":
|
||
return {
|
||
"email": decision.email,
|
||
"token": decision.token,
|
||
"verification_url": decision.verification_url,
|
||
"message": decision.message,
|
||
}
|
||
if decision.message:
|
||
raise ValueError(decision.message)
|
||
raise RuntimeError("resend_failed")
|
||
|
||
|
||
class GatewayUserStore:
|
||
def __init__(self, db_path: str | Path | None = None):
|
||
self.db_path = Path(db_path or default_db_path())
|
||
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
||
self._init_db()
|
||
self._verification = _VerificationStateMachine(self)
|
||
|
||
@property
|
||
def verification(self) -> _VerificationStateMachine:
|
||
return self._verification
|
||
|
||
def _connect(self) -> sqlite3.Connection:
|
||
conn = sqlite3.connect(self.db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA foreign_keys = ON")
|
||
return conn
|
||
|
||
def _init_db(self) -> None:
|
||
with self._connect() as conn:
|
||
conn.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS principals (
|
||
principal_id TEXT PRIMARY KEY,
|
||
email TEXT NOT NULL UNIQUE,
|
||
status TEXT NOT NULL DEFAULT 'verified',
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS external_identities (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
principal_id TEXT NOT NULL,
|
||
platform TEXT NOT NULL,
|
||
external_user_id TEXT NOT NULL,
|
||
user_name TEXT,
|
||
verified_email TEXT NOT NULL,
|
||
verified_at TEXT NOT NULL,
|
||
UNIQUE(platform, external_user_id),
|
||
UNIQUE(platform, verified_email),
|
||
FOREIGN KEY(principal_id) REFERENCES principals(principal_id)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS identity_enforcement (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
platform TEXT NOT NULL,
|
||
external_user_id TEXT NOT NULL,
|
||
external_user_name TEXT,
|
||
state TEXT NOT NULL DEFAULT 'new',
|
||
principal_id TEXT,
|
||
verified_email TEXT,
|
||
pending_email TEXT,
|
||
failure_cycles INTEGER NOT NULL DEFAULT 0,
|
||
blocked_until TEXT,
|
||
last_request_id INTEGER,
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL,
|
||
UNIQUE(platform, external_user_id)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS verification_requests (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
platform TEXT NOT NULL,
|
||
external_user_id TEXT NOT NULL,
|
||
email TEXT NOT NULL,
|
||
token_hash TEXT NOT NULL UNIQUE,
|
||
status TEXT NOT NULL,
|
||
expires_at TEXT NOT NULL,
|
||
verified_at TEXT,
|
||
created_at TEXT NOT NULL,
|
||
last_sent_at TEXT NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS quota_policies (
|
||
principal_id TEXT PRIMARY KEY,
|
||
daily_message_limit INTEGER,
|
||
daily_token_limit INTEGER,
|
||
notes TEXT,
|
||
updated_at TEXT NOT NULL,
|
||
FOREIGN KEY(principal_id) REFERENCES principals(principal_id)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS model_policies (
|
||
principal_id TEXT PRIMARY KEY,
|
||
allowed_models_json TEXT,
|
||
default_model TEXT,
|
||
updated_at TEXT NOT NULL,
|
||
FOREIGN KEY(principal_id) REFERENCES principals(principal_id)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS admin_audit_logs (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
actor TEXT NOT NULL,
|
||
action TEXT NOT NULL,
|
||
platform TEXT,
|
||
external_user_id TEXT,
|
||
details_json TEXT,
|
||
created_at TEXT NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS dashboard_roles (
|
||
email TEXT PRIMARY KEY,
|
||
role TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL,
|
||
updated_by TEXT NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS principal_usage_daily (
|
||
principal_id TEXT NOT NULL,
|
||
usage_date TEXT NOT NULL,
|
||
message_count INTEGER NOT NULL DEFAULT 0,
|
||
input_tokens INTEGER NOT NULL DEFAULT 0,
|
||
output_tokens INTEGER NOT NULL DEFAULT 0,
|
||
total_tokens INTEGER NOT NULL DEFAULT 0,
|
||
last_model TEXT,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (principal_id, usage_date),
|
||
FOREIGN KEY(principal_id) REFERENCES principals(principal_id)
|
||
);
|
||
"""
|
||
)
|
||
|
||
@staticmethod
|
||
def supports_source(source: SessionSource | None) -> bool:
|
||
identity_key = _source_identity_key(source)
|
||
return bool(
|
||
source
|
||
and source.chat_type == "dm"
|
||
and getattr(source.platform, "value", source.platform) in _SUPPORTED_PLATFORM_VALUES
|
||
and identity_key
|
||
)
|
||
|
||
def _audit(
|
||
self,
|
||
conn: sqlite3.Connection,
|
||
*,
|
||
actor: str,
|
||
action: str,
|
||
platform: str | None,
|
||
external_user_id: str | None,
|
||
details: dict[str, Any],
|
||
) -> None:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO admin_audit_logs(actor, action, platform, external_user_id, details_json, created_at)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(actor, action, platform, external_user_id, json.dumps(details, ensure_ascii=False), _iso(_utcnow())),
|
||
)
|
||
|
||
def _get_identity_row(self, conn: sqlite3.Connection, platform: str, external_user_id: str) -> sqlite3.Row | None:
|
||
return conn.execute(
|
||
"SELECT * FROM identity_enforcement WHERE platform = ? AND external_user_id = ?",
|
||
(platform, external_user_id),
|
||
).fetchone()
|
||
|
||
def _build_principal_context(
|
||
self,
|
||
conn: sqlite3.Connection,
|
||
*,
|
||
principal_id: str,
|
||
identity: dict[str, Any],
|
||
) -> dict[str, Any] | None:
|
||
principal = conn.execute("SELECT * FROM principals WHERE principal_id = ?", (principal_id,)).fetchone()
|
||
if principal is None:
|
||
return None
|
||
quota_policy = conn.execute("SELECT * FROM quota_policies WHERE principal_id = ?", (principal_id,)).fetchone()
|
||
model_policy = conn.execute("SELECT * FROM model_policies WHERE principal_id = ?", (principal_id,)).fetchone()
|
||
usage = conn.execute(
|
||
"SELECT * FROM principal_usage_daily WHERE principal_id = ? AND usage_date = ?",
|
||
(principal_id, _local_day_key()),
|
||
).fetchone()
|
||
model_item = dict(model_policy) if model_policy is not None else None
|
||
if model_item is not None:
|
||
model_item["allowed_models"] = json.loads(model_item.pop("allowed_models_json") or "[]")
|
||
return {
|
||
"principal_id": principal_id,
|
||
"principal": dict(principal),
|
||
"identity": identity,
|
||
"quota_policy": dict(quota_policy) if quota_policy is not None else None,
|
||
"model_policy": model_item,
|
||
"usage": dict(usage) if usage is not None else None,
|
||
}
|
||
|
||
def _rate_limit_message(self, *, seconds: int | None = None, window_minutes: int | None = None, max_sends: int | None = None) -> str:
|
||
if seconds is not None:
|
||
return f"驗證信剛寄出，請稍候 {seconds} 秒後再重試。"
|
||
return f"此 Email 在 {window_minutes} 分鐘內最多只能寄送 {max_sends} 封驗證信，請稍後再試。"
|
||
|
||
def _check_send_rate_limits(
|
||
self,
|
||
conn: sqlite3.Connection,
|
||
*,
|
||
platform: str,
|
||
external_user_id: str,
|
||
email: str,
|
||
now: datetime,
|
||
) -> str | None:
|
||
identity_recent = conn.execute(
|
||
"""
|
||
SELECT last_sent_at FROM verification_requests
|
||
WHERE platform = ? AND external_user_id = ?
|
||
ORDER BY id DESC LIMIT 1
|
||
""",
|
||
(platform, external_user_id),
|
||
).fetchone()
|
||
if identity_recent is not None:
|
||
last_sent_at = _parse_iso(identity_recent["last_sent_at"])
|
||
if last_sent_at and (now - last_sent_at).total_seconds() < _SEND_COOLDOWN_SECONDS:
|
||
return self._rate_limit_message(seconds=_SEND_COOLDOWN_SECONDS)
|
||
|
||
window_start = _iso(now - timedelta(minutes=_EMAIL_WINDOW_MINUTES))
|
||
email_count = conn.execute(
|
||
"""
|
||
SELECT COUNT(1) AS c FROM verification_requests
|
||
WHERE email = ? AND last_sent_at >= ?
|
||
""",
|
||
(email, window_start),
|
||
).fetchone()
|
||
if email_count and int(email_count["c"] or 0) >= _EMAIL_WINDOW_MAX_SENDS:
|
||
return self._rate_limit_message(
|
||
window_minutes=_EMAIL_WINDOW_MINUTES,
|
||
max_sends=_EMAIL_WINDOW_MAX_SENDS,
|
||
)
|
||
return None
|
||
|
||
def _ensure_identity_row(self, conn: sqlite3.Connection, source: SessionSource) -> sqlite3.Row:
|
||
platform = source.platform.value
|
||
user_id = str(source.user_id)
|
||
row = self._get_identity_row(conn, platform, user_id)
|
||
now = _iso(_utcnow())
|
||
if row is None:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO identity_enforcement(
|
||
platform, external_user_id, external_user_name, state, created_at, updated_at
|
||
) VALUES (?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(platform, user_id, source.user_name, _STATE_NEW, now, now),
|
||
)
|
||
row = self._get_identity_row(conn, platform, user_id)
|
||
elif source.user_name and source.user_name != row["external_user_name"]:
|
||
conn.execute(
|
||
"UPDATE identity_enforcement SET external_user_name = ?, updated_at = ? WHERE id = ?",
|
||
(source.user_name, now, row["id"]),
|
||
)
|
||
row = self._get_identity_row(conn, platform, user_id)
|
||
if row is None:
|
||
raise RuntimeError("failed to create identity_enforcement row")
|
||
return row
|
||
|
||
def _reconcile_identity_state(self, conn: sqlite3.Connection, source: SessionSource) -> sqlite3.Row:
|
||
row = self._ensure_identity_row(conn, source)
|
||
state = row["state"]
|
||
now = _utcnow()
|
||
blocked_until = _parse_iso(row["blocked_until"])
|
||
updated = False
|
||
|
||
if state == _STATE_PENDING and row["last_request_id"]:
|
||
req = conn.execute(
|
||
"SELECT * FROM verification_requests WHERE id = ?",
|
||
(row["last_request_id"],),
|
||
).fetchone()
|
||
expires_at = _parse_iso(req["expires_at"] if req else None)
|
||
if req and req["status"] == _STATE_PENDING and expires_at and expires_at <= now:
|
||
conn.execute(
|
||
"UPDATE verification_requests SET status = 'expired' WHERE id = ?",
|
||
(req["id"],),
|
||
)
|
||
failure_cycles = int(row["failure_cycles"] or 0) + 1
|
||
next_state = _STATE_PERMANENT if failure_cycles >= 3 else _STATE_COOLDOWN
|
||
next_blocked = None if next_state == _STATE_PERMANENT else _iso(now + timedelta(hours=72))
|
||
conn.execute(
|
||
"""
|
||
UPDATE identity_enforcement
|
||
SET state = ?, failure_cycles = ?, blocked_until = ?, pending_email = NULL, updated_at = ?
|
||
WHERE id = ?
|
||
""",
|
||
(next_state, failure_cycles, next_blocked, _iso(now), row["id"]),
|
||
)
|
||
updated = True
|
||
|
||
elif state == _STATE_COOLDOWN and blocked_until and blocked_until <= now:
|
||
conn.execute(
|
||
"""
|
||
UPDATE identity_enforcement
|
||
SET state = ?, blocked_until = NULL, pending_email = NULL, updated_at = ?
|
||
WHERE id = ?
|
||
""",
|
||
(_STATE_NEW, _iso(now), row["id"]),
|
||
)
|
||
updated = True
|
||
|
||
if updated:
|
||
row = self._get_identity_row(conn, source.platform.value, str(source.user_id))
|
||
if row is None:
|
||
raise RuntimeError("identity_enforcement row disappeared during reconciliation")
|
||
return row
|
||
|
||
def is_verified_source(self, source: SessionSource) -> bool:
|
||
# Verification is intentionally two-part: the runtime state row must say
|
||
# VERIFIED *and* the binding row must still exist. This prevents stale
|
||
# `identity_enforcement` rows from authorizing a source whose binding was
|
||
# removed later. For the full LINE/Telegram lifecycle and common regressions
|
||
# (unbind without pairing revoke, onboarding fail-closed, stale email handoff),
|
||
# see:
|
||
# skills/devops/gateway-identity-binding-lifecycle/references/
|
||
# line-telegram-verification-lifecycle.md
|
||
if not self.supports_source(source):
|
||
return False
|
||
with self._connect() as conn:
|
||
row = self._get_identity_row(conn, source.platform.value, str(source.user_id))
|
||
if row is None:
|
||
return False
|
||
if row["state"] != _STATE_VERIFIED:
|
||
return False
|
||
identity = conn.execute(
|
||
"SELECT 1 FROM external_identities WHERE platform = ? AND external_user_id = ?",
|
||
(source.platform.value, str(source.user_id)),
|
||
).fetchone()
|
||
return identity is not None
|
||
|
||
def get_verified_email_for_source(self, source: SessionSource) -> str | None:
|
||
"""Return the bound verified email for a verified source, if any."""
|
||
# Keep this lookup aligned with `is_verified_source()`: email handoff must
|
||
# only trust live bindings that still exist in both `external_identities`
|
||
# and `identity_enforcement`. Do not replace this with a principals-only
|
||
# lookup for LINE/Telegram DM sources.
|
||
if not self.supports_source(source):
|
||
return None
|
||
platform_value = getattr(source.platform, "value", source.platform)
|
||
identity_key = _source_identity_key(source)
|
||
with self._connect() as conn:
|
||
row = conn.execute(
|
||
"""
|
||
SELECT e.verified_email
|
||
FROM external_identities e
|
||
JOIN identity_enforcement i
|
||
ON i.platform = e.platform
|
||
AND i.external_user_id = e.external_user_id
|
||
WHERE e.platform = ?
|
||
AND e.external_user_id = ?
|
||
AND i.state = ?
|
||
""",
|
||
(platform_value, identity_key, _STATE_VERIFIED),
|
||
).fetchone()
|
||
if row is not None:
|
||
email = str(row["verified_email"] or "").strip()
|
||
return email or None
|
||
if platform_value == "email":
|
||
row = conn.execute(
|
||
"SELECT email FROM principals WHERE email = ? AND status = 'verified'",
|
||
(identity_key,),
|
||
).fetchone()
|
||
if row is not None:
|
||
email = str(row["email"] or "").strip()
|
||
return email or None
|
||
return None
|
||
|
||
def bound_email_matches_source(
|
||
self,
|
||
source: SessionSource,
|
||
target_email: str | None,
|
||
*,
|
||
explicit_user_requested: bool = False,
|
||
) -> bool:
|
||
"""True when a mail handoff target is allowed for this source.
|
||
|
||
Use this immediately before any gateway mail handoff so a stale prompt,
|
||
cached session field, or compressed context cannot redirect delivery to a
|
||
different principal than the one currently bound to the active source.
|
||
|
||
When the user explicitly asked to send to some other email address, pass
|
||
``explicit_user_requested=True`` to bypass the bound-email equality
|
||
check. The strict match is only for *system-initiated* delivery to the
|
||
source's own bound mailbox.
|
||
"""
|
||
if explicit_user_requested:
|
||
return bool(str(target_email or "").strip())
|
||
expected = str(self.get_verified_email_for_source(source) or "").strip().lower()
|
||
actual = str(target_email or "").strip().lower()
|
||
return bool(expected and actual and expected == actual)
|
||
|
||
def process_inbound_message(
|
||
self,
|
||
source: SessionSource,
|
||
text: str | None,
|
||
*,
|
||
public_base_url: Optional[str] = None,
|
||
) -> VerificationDecision:
|
||
# Adapter intake must leave room for this method to run for new DM users;
|
||
# otherwise Telegram/LINE onboarding silently dies before company-email
|
||
# verification can start. See the lifecycle note at:
|
||
# skills/devops/gateway-identity-binding-lifecycle/references/
|
||
# line-telegram-verification-lifecycle.md
|
||
return self._verification.process_inbound_message(
|
||
source,
|
||
text,
|
||
public_base_url=public_base_url,
|
||
)
|
||
|
||
def verify_token(self, token: str) -> VerificationResult:
|
||
return self._verification.verify_token(token)
|
||
|
||
def list_identities(self, *, state: Optional[str] = None) -> list[dict[str, Any]]:
|
||
with self._connect() as conn:
|
||
sql = "SELECT * FROM identity_enforcement"
|
||
params: tuple[Any, ...] = ()
|
||
if state:
|
||
sql += " WHERE state = ?"
|
||
params = (state,)
|
||
sql += " ORDER BY updated_at DESC, id DESC"
|
||
rows = conn.execute(sql, params).fetchall()
|
||
return [dict(r) for r in rows]
|
||
|
||
def list_principals(self) -> list[dict[str, Any]]:
|
||
with self._connect() as conn:
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT p.principal_id, p.email, p.status, p.created_at, p.updated_at,
|
||
COALESCE(json_group_array(json_object(
|
||
'platform', e.platform,
|
||
'external_user_id', e.external_user_id,
|
||
'verified_at', e.verified_at
|
||
)), '[]') AS identities_json
|
||
FROM principals p
|
||
LEFT JOIN external_identities e ON e.principal_id = p.principal_id
|
||
GROUP BY p.principal_id
|
||
ORDER BY p.updated_at DESC
|
||
"""
|
||
).fetchall()
|
||
result: list[dict[str, Any]] = []
|
||
for row in rows:
|
||
item = dict(row)
|
||
item["identities"] = json.loads(item.pop("identities_json") or "[]")
|
||
result.append(item)
|
||
return result
|
||
|
||
def list_bound_identities_for_email(self, email: str | None) -> list[dict[str, Any]]:
|
||
normalized = _normalize_email(email or "")
|
||
if not normalized:
|
||
return []
|
||
with self._connect() as conn:
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT principal_id, platform, external_user_id, user_name, verified_email, verified_at
|
||
FROM external_identities
|
||
WHERE verified_email = ?
|
||
ORDER BY platform, external_user_id
|
||
""",
|
||
(normalized,),
|
||
).fetchall()
|
||
return [dict(r) for r in rows]
|
||
|
||
def unblock_identity(self, platform: str, external_user_id: str, *, actor: str = "admin") -> dict[str, Any]:
|
||
with self._connect() as conn:
|
||
row = self._get_identity_row(conn, platform, external_user_id)
|
||
if row is None:
|
||
raise KeyError("identity_not_found")
|
||
now_iso = _iso(_utcnow())
|
||
conn.execute(
|
||
"""
|
||
UPDATE identity_enforcement
|
||
SET state = CASE WHEN verified_email IS NOT NULL THEN ? ELSE ? END,
|
||
blocked_until = NULL,
|
||
pending_email = NULL,
|
||
updated_at = ?
|
||
WHERE id = ?
|
||
""",
|
||
(_STATE_VERIFIED, _STATE_NEW, now_iso, row["id"]),
|
||
)
|
||
self._audit(conn, actor=actor, action="unblock_identity", platform=platform, external_user_id=external_user_id, details={})
|
||
updated = self._get_identity_row(conn, platform, external_user_id)
|
||
if updated is None:
|
||
raise RuntimeError("identity_not_found_after_unblock")
|
||
return dict(updated)
|
||
|
||
def admin_block_identity(self, platform: str, external_user_id: str, *, actor: str = "admin") -> dict[str, Any]:
|
||
with self._connect() as conn:
|
||
row = self._get_identity_row(conn, platform, external_user_id)
|
||
if row is None:
|
||
raise KeyError("identity_not_found")
|
||
conn.execute(
|
||
"UPDATE identity_enforcement SET state = ?, blocked_until = NULL, updated_at = ? WHERE id = ?",
|
||
(_STATE_ADMIN_BLOCKED, _iso(_utcnow()), row["id"]),
|
||
)
|
||
self._audit(conn, actor=actor, action="admin_block_identity", platform=platform, external_user_id=external_user_id, details={})
|
||
updated = self._get_identity_row(conn, platform, external_user_id)
|
||
if updated is None:
|
||
raise RuntimeError("identity_not_found_after_block")
|
||
return dict(updated)
|
||
|
||
def force_bind_identity(
|
||
self,
|
||
*,
|
||
platform: str,
|
||
external_user_id: str,
|
||
email: str,
|
||
user_name: Optional[str] = None,
|
||
actor: str = "admin",
|
||
) -> dict[str, Any]:
|
||
email = _normalize_email(email)
|
||
if not _is_allowed_domain(email):
|
||
raise ValueError("invalid_domain")
|
||
with self._connect() as conn:
|
||
existing = conn.execute(
|
||
"SELECT external_user_id FROM external_identities WHERE platform = ? AND verified_email = ?",
|
||
(platform, email),
|
||
).fetchone()
|
||
if existing and existing["external_user_id"] != external_user_id:
|
||
raise ValueError("email_already_bound")
|
||
principal = conn.execute("SELECT * FROM principals WHERE email = ?", (email,)).fetchone()
|
||
now = _iso(_utcnow())
|
||
principal_id = principal["principal_id"] if principal else secrets.token_hex(12)
|
||
if principal is None:
|
||
conn.execute(
|
||
"INSERT INTO principals(principal_id, email, status, created_at, updated_at) VALUES (?, ?, 'verified', ?, ?)",
|
||
(principal_id, email, now, now),
|
||
)
|
||
conn.execute(
|
||
"DELETE FROM external_identities WHERE platform = ? AND external_user_id = ?",
|
||
(platform, external_user_id),
|
||
)
|
||
conn.execute(
|
||
"DELETE FROM external_identities WHERE platform = ? AND verified_email = ? AND external_user_id != ?",
|
||
(platform, email, external_user_id),
|
||
)
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO external_identities(
|
||
principal_id, platform, external_user_id, user_name, verified_email, verified_at
|
||
) VALUES (?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(principal_id, platform, external_user_id, user_name, email, now),
|
||
)
|
||
source = SessionSource(platform=Platform(platform), chat_id=external_user_id, user_id=external_user_id, user_name=user_name, chat_type="dm")
|
||
row = self._ensure_identity_row(conn, source)
|
||
conn.execute(
|
||
"""
|
||
UPDATE identity_enforcement
|
||
SET state = ?, principal_id = ?, verified_email = ?, pending_email = NULL,
|
||
blocked_until = NULL, updated_at = ?
|
||
WHERE id = ?
|
||
""",
|
||
(_STATE_VERIFIED, principal_id, email, now, row["id"]),
|
||
)
|
||
self._audit(conn, actor=actor, action="force_bind_identity", platform=platform, external_user_id=external_user_id, details={"email": email})
|
||
updated = self._get_identity_row(conn, platform, external_user_id)
|
||
if updated is None:
|
||
raise RuntimeError("identity_not_found_after_force_bind")
|
||
return dict(updated)
|
||
|
||
def unbind_identity(self, platform: str, external_user_id: str, *, actor: str = "admin") -> dict[str, Any]:
|
||
# Unbind is a multi-surface revoke, not a single-row delete. Runtime DB
|
||
# cleanup alone is insufficient because gateway authz also honors pairing
|
||
# approvals as an independent grant. Keep the DB reset + PairingStore.revoke
|
||
# in sync unless the architecture is deliberately redesigned; otherwise the
|
||
# classic "admin shows unbound but user can still chat" regression returns.
|
||
# Full background:
|
||
# skills/devops/gateway-identity-binding-lifecycle/references/
|
||
# line-telegram-verification-lifecycle.md
|
||
pairing_revoked = False
|
||
with self._connect() as conn:
|
||
row = self._get_identity_row(conn, platform, external_user_id)
|
||
if row is None:
|
||
raise KeyError("identity_not_found")
|
||
conn.execute(
|
||
"DELETE FROM external_identities WHERE platform = ? AND external_user_id = ?",
|
||
(platform, external_user_id),
|
||
)
|
||
conn.execute(
|
||
"""
|
||
UPDATE identity_enforcement
|
||
SET state = ?, principal_id = NULL, verified_email = NULL, pending_email = NULL,
|
||
blocked_until = NULL, updated_at = ?
|
||
WHERE id = ?
|
||
""",
|
||
(_STATE_NEW, _iso(_utcnow()), row["id"]),
|
||
)
|
||
pairing_revoked = PairingStore().revoke(platform, external_user_id)
|
||
self._audit(
|
||
conn,
|
||
actor=actor,
|
||
action="unbind_identity",
|
||
platform=platform,
|
||
external_user_id=external_user_id,
|
||
details={"pairing_revoked": pairing_revoked},
|
||
)
|
||
updated = self._get_identity_row(conn, platform, external_user_id)
|
||
if updated is None:
|
||
raise RuntimeError("identity_not_found_after_unbind")
|
||
return dict(updated)
|
||
|
||
def resend_verification(self, platform: str, external_user_id: str, *, public_base_url: Optional[str] = None) -> dict[str, Any]:
|
||
return self._verification.resend_verification(
|
||
platform,
|
||
external_user_id,
|
||
public_base_url=public_base_url,
|
||
)
|
||
|
||
def get_dashboard_role(self, email: str | None) -> str | None:
|
||
normalized = _normalize_email(email or "")
|
||
if not normalized:
|
||
return None
|
||
with self._connect() as conn:
|
||
row = conn.execute("SELECT role FROM dashboard_roles WHERE email = ?", (normalized,)).fetchone()
|
||
return str(row["role"]) if row is not None else None
|
||
|
||
def list_dashboard_roles(self) -> list[dict[str, Any]]:
|
||
with self._connect() as conn:
|
||
return [dict(r) for r in conn.execute("SELECT * FROM dashboard_roles ORDER BY email ASC").fetchall()]
|
||
|
||
def set_dashboard_role(self, email: str, role: str, *, actor: str = "admin") -> dict[str, Any]:
|
||
normalized = _normalize_email(email)
|
||
normalized_role = (role or "").strip().lower()
|
||
if normalized_role not in _VALID_ROLES:
|
||
raise ValueError("invalid_role")
|
||
with self._connect() as conn:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO dashboard_roles(email, role, updated_at, updated_by)
|
||
VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(email) DO UPDATE SET
|
||
role = excluded.role,
|
||
updated_at = excluded.updated_at,
|
||
updated_by = excluded.updated_by
|
||
""",
|
||
(normalized, normalized_role, _iso(_utcnow()), actor),
|
||
)
|
||
row = conn.execute("SELECT * FROM dashboard_roles WHERE email = ?", (normalized,)).fetchone()
|
||
if row is None:
|
||
raise RuntimeError("dashboard_role_not_persisted")
|
||
return dict(row)
|
||
|
||
def get_principal_usage(self, principal_id: str, *, usage_date: str | None = None) -> dict[str, Any] | None:
|
||
day_key = usage_date or _local_day_key()
|
||
with self._connect() as conn:
|
||
row = conn.execute(
|
||
"SELECT * FROM principal_usage_daily WHERE principal_id = ? AND usage_date = ?",
|
||
(principal_id, day_key),
|
||
).fetchone()
|
||
return dict(row) if row is not None else None
|
||
|
||
def list_principal_usage(self, *, limit: int = 100, usage_date: str | None = None) -> list[dict[str, Any]]:
|
||
day_key = usage_date or _local_day_key()
|
||
with self._connect() as conn:
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT
|
||
usage.principal_id,
|
||
usage.usage_date,
|
||
usage.message_count,
|
||
usage.input_tokens,
|
||
usage.output_tokens,
|
||
usage.total_tokens,
|
||
usage.last_model,
|
||
usage.updated_at,
|
||
principals.email,
|
||
quota.daily_message_limit,
|
||
quota.daily_token_limit,
|
||
quota.notes AS quota_notes
|
||
FROM principal_usage_daily AS usage
|
||
LEFT JOIN principals ON principals.principal_id = usage.principal_id
|
||
LEFT JOIN quota_policies AS quota ON quota.principal_id = usage.principal_id
|
||
WHERE usage.usage_date = ?
|
||
ORDER BY usage.total_tokens DESC, usage.message_count DESC, usage.updated_at DESC
|
||
LIMIT ?
|
||
""",
|
||
(day_key, max(1, min(limit, 500))),
|
||
).fetchall()
|
||
return [dict(row) for row in rows]
|
||
|
||
def increment_principal_usage(
|
||
self,
|
||
principal_id: str,
|
||
*,
|
||
input_tokens: int,
|
||
output_tokens: int,
|
||
message_count: int = 1,
|
||
model: str | None = None,
|
||
) -> dict[str, Any]:
|
||
day_key = _local_day_key()
|
||
total_tokens = max(0, int(input_tokens or 0)) + max(0, int(output_tokens or 0))
|
||
with self._connect() as conn:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO principal_usage_daily(
|
||
principal_id, usage_date, message_count, input_tokens, output_tokens, total_tokens, last_model, updated_at
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(principal_id, usage_date) DO UPDATE SET
|
||
message_count = principal_usage_daily.message_count + excluded.message_count,
|
||
input_tokens = principal_usage_daily.input_tokens + excluded.input_tokens,
|
||
output_tokens = principal_usage_daily.output_tokens + excluded.output_tokens,
|
||
total_tokens = principal_usage_daily.total_tokens + excluded.total_tokens,
|
||
last_model = excluded.last_model,
|
||
updated_at = excluded.updated_at
|
||
""",
|
||
(
|
||
principal_id,
|
||
day_key,
|
||
max(0, int(message_count or 0)),
|
||
max(0, int(input_tokens or 0)),
|
||
max(0, int(output_tokens or 0)),
|
||
total_tokens,
|
||
model,
|
||
_iso(_utcnow()),
|
||
),
|
||
)
|
||
row = conn.execute(
|
||
"SELECT * FROM principal_usage_daily WHERE principal_id = ? AND usage_date = ?",
|
||
(principal_id, day_key),
|
||
).fetchone()
|
||
if row is None:
|
||
raise RuntimeError("principal_usage_not_persisted")
|
||
return dict(row)
|
||
|
||
def record_usage_for_source(
|
||
self,
|
||
source: SessionSource | None,
|
||
*,
|
||
input_tokens: int,
|
||
output_tokens: int,
|
||
total_tokens: int | None = None,
|
||
message_count: int = 1,
|
||
model: str | None = None,
|
||
) -> dict[str, Any] | None:
|
||
if not self.supports_source(source):
|
||
return None
|
||
context = self.get_principal_context(source)
|
||
if context is None or not context.get("principal_id"):
|
||
return None
|
||
input_value = max(0, int(input_tokens or 0))
|
||
output_value = max(0, int(output_tokens or 0))
|
||
total_value = max(0, int(total_tokens or 0))
|
||
if input_value == 0 and output_value == 0 and total_value > 0:
|
||
input_value = total_value
|
||
if input_value == 0 and output_value == 0 and max(0, int(message_count or 0)) <= 0:
|
||
return None
|
||
return self.increment_principal_usage(
|
||
str(context["principal_id"]),
|
||
input_tokens=input_value,
|
||
output_tokens=output_value,
|
||
message_count=message_count,
|
||
model=model,
|
||
)
|
||
|
||
def get_principal_context(self, source: SessionSource | None) -> dict[str, Any] | None:
|
||
if not self.supports_source(source):
|
||
return None
|
||
platform_value = str(getattr(getattr(source, "platform", None), "value", getattr(source, "platform", "")) or "")
|
||
identity_key = _source_identity_key(source)
|
||
with self._connect() as conn:
|
||
row = self._get_identity_row(conn, platform_value, identity_key)
|
||
if row is not None and row["state"] == _STATE_VERIFIED and row["principal_id"]:
|
||
return self._build_principal_context(
|
||
conn,
|
||
principal_id=str(row["principal_id"]),
|
||
identity=dict(row),
|
||
)
|
||
if platform_value == "email":
|
||
principal = conn.execute(
|
||
"SELECT * FROM principals WHERE email = ? AND status = 'verified'",
|
||
(identity_key,),
|
||
).fetchone()
|
||
if principal is None:
|
||
return None
|
||
return self._build_principal_context(
|
||
conn,
|
||
principal_id=str(principal["principal_id"]),
|
||
identity={
|
||
"platform": "email",
|
||
"external_user_id": identity_key,
|
||
"state": _STATE_VERIFIED,
|
||
"principal_id": str(principal["principal_id"]),
|
||
"verified_email": str(principal["email"] or identity_key),
|
||
},
|
||
)
|
||
return None
|
||
|
||
def get_quota_policies(self) -> list[dict[str, Any]]:
|
||
with self._connect() as conn:
|
||
return [dict(r) for r in conn.execute("SELECT * FROM quota_policies ORDER BY updated_at DESC").fetchall()]
|
||
|
||
def upsert_quota_policy(self, principal_id: str, *, daily_message_limit: Optional[int], daily_token_limit: Optional[int], notes: str = "") -> dict[str, Any]:
|
||
with self._connect() as conn:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO quota_policies(principal_id, daily_message_limit, daily_token_limit, notes, updated_at)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
ON CONFLICT(principal_id) DO UPDATE SET
|
||
daily_message_limit = excluded.daily_message_limit,
|
||
daily_token_limit = excluded.daily_token_limit,
|
||
notes = excluded.notes,
|
||
updated_at = excluded.updated_at
|
||
""",
|
||
(principal_id, daily_message_limit, daily_token_limit, notes, _iso(_utcnow())),
|
||
)
|
||
row = conn.execute("SELECT * FROM quota_policies WHERE principal_id = ?", (principal_id,)).fetchone()
|
||
return dict(row)
|
||
|
||
def get_model_policies(self) -> list[dict[str, Any]]:
|
||
with self._connect() as conn:
|
||
rows = conn.execute("SELECT * FROM model_policies ORDER BY updated_at DESC").fetchall()
|
||
result = []
|
||
for row in rows:
|
||
item = dict(row)
|
||
item["allowed_models"] = json.loads(item.pop("allowed_models_json") or "[]")
|
||
result.append(item)
|
||
return result
|
||
|
||
def upsert_model_policy(self, principal_id: str, *, allowed_models: list[str], default_model: str | None) -> dict[str, Any]:
|
||
with self._connect() as conn:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO model_policies(principal_id, allowed_models_json, default_model, updated_at)
|
||
VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(principal_id) DO UPDATE SET
|
||
allowed_models_json = excluded.allowed_models_json,
|
||
default_model = excluded.default_model,
|
||
updated_at = excluded.updated_at
|
||
""",
|
||
(principal_id, json.dumps(allowed_models, ensure_ascii=False), default_model, _iso(_utcnow())),
|
||
)
|
||
row = conn.execute("SELECT * FROM model_policies WHERE principal_id = ?", (principal_id,)).fetchone()
|
||
item = dict(row)
|
||
item["allowed_models"] = json.loads(item.pop("allowed_models_json") or "[]")
|
||
return item
|
||
|
||
def list_audit_logs(self, *, limit: int = 100) -> list[dict[str, Any]]:
|
||
with self._connect() as conn:
|
||
rows = conn.execute(
|
||
"SELECT * FROM admin_audit_logs ORDER BY id DESC LIMIT ?",
|
||
(max(1, min(limit, 500)),),
|
||
).fetchall()
|
||
result = []
|
||
for row in rows:
|
||
item = dict(row)
|
||
item["details"] = json.loads(item.pop("details_json") or "{}")
|
||
result.append(item)
|
||
return result
|