b279d6c61a
- core/telemetry.py is now a no-op (no opentelemetry imports); entrypoints no longer call setup/shutdown telemetry - Remove all opentelemetry-* deps from pyproject groups; regenerate uv.lock - Drop otel_exporter_otlp_endpoint/otel_service_name settings; sentry and auth use "contract-check" instead - Stop publishing worker metrics ports; bind API metrics to 127.0.0.1 by default via API_METRICS_BIND_HOST - Replace otel-collector with Vector in observer profile (docker_logs + prometheus_scrape -> OpenObserve); add deploy/observability/vector-config.yaml - Update .env.example, docs (ARCHITECTURE, DEPLOY), README, Makefile - Rewrite tests/unit/test_telemetry.py for the no-op implementation
349 lines
11 KiB
Python
349 lines
11 KiB
Python
"""User authentication helpers: JWT signing/verification and Telegram identity checks.
|
|
|
|
This module is part of `core` and may be imported by the API. It does NOT depend
|
|
on DB/S3/MQ — only on pydantic-settings + python-jose-style JWT via PyJWT +
|
|
standard library hmac/hashlib.
|
|
|
|
Supports three Telegram identity sources:
|
|
1. Bot adapter: the bot already got a verified `telegram_id` from Telegram and
|
|
uses a service token to exchange it for a user JWT.
|
|
2. Telegram Login Widget: web callback payload signed by Telegram.
|
|
3. Telegram Mini App: `initData` signed by Telegram.
|
|
|
|
All sources issue the same JWT containing `sub` (user UUID) and `telegram_id`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from urllib.parse import parse_qsl
|
|
|
|
import jwt
|
|
|
|
from src.contract_check.core.config import get_settings
|
|
from src.contract_check.core.logging import get_logger
|
|
|
|
log = get_logger(__name__)
|
|
|
|
JWT_TYPE_ACCESS = "access"
|
|
JWT_TYPE_REFRESH = "refresh"
|
|
|
|
|
|
class AuthError(Exception):
|
|
"""Raised when an identity proof cannot be verified or a token is invalid."""
|
|
|
|
|
|
class TokenExpiredError(AuthError):
|
|
"""JWT has expired."""
|
|
|
|
|
|
class TokenInvalidError(AuthError):
|
|
"""JWT is malformed or signature is bad."""
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class UserIdentity:
|
|
"""Verified user identity returned by Telegram identity checks."""
|
|
|
|
telegram_id: int
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AccessTokenClaims:
|
|
"""Payload we put into (and expect from) an access JWT."""
|
|
|
|
sub: uuid.UUID # user_id
|
|
telegram_id: int
|
|
type: str
|
|
exp: int | None = None # unix seconds
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"sub": str(self.sub),
|
|
"telegram_id": self.telegram_id,
|
|
"type": self.type,
|
|
"exp": self.exp,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, payload: dict[str, Any]) -> AccessTokenClaims:
|
|
return cls(
|
|
sub=uuid.UUID(str(payload["sub"])),
|
|
telegram_id=int(payload["telegram_id"]),
|
|
type=str(payload.get("type", JWT_TYPE_ACCESS)),
|
|
exp=payload.get("exp"),
|
|
)
|
|
|
|
|
|
def _jwt_secret() -> str:
|
|
return get_settings().jwt_secret
|
|
|
|
|
|
def _jwt_algorithm() -> str:
|
|
return get_settings().jwt_algorithm
|
|
|
|
|
|
def _jwt_access_ttl() -> dt.timedelta:
|
|
return dt.timedelta(minutes=get_settings().jwt_access_ttl_minutes)
|
|
|
|
|
|
def _jwt_refresh_ttl() -> dt.timedelta:
|
|
return dt.timedelta(days=get_settings().jwt_refresh_ttl_days)
|
|
|
|
|
|
def create_access_token(user_id: uuid.UUID, telegram_id: int) -> str:
|
|
"""Sign a fresh access JWT for a verified user."""
|
|
now = dt.datetime.now(tz=dt.UTC)
|
|
claims = AccessTokenClaims(
|
|
sub=user_id,
|
|
telegram_id=telegram_id,
|
|
type=JWT_TYPE_ACCESS,
|
|
)
|
|
payload = claims.to_dict()
|
|
payload.update(
|
|
{
|
|
"iat": int(now.timestamp()),
|
|
"exp": int((now + _jwt_access_ttl()).timestamp()),
|
|
"iss": "contract-check",
|
|
"aud": "contract-check",
|
|
}
|
|
)
|
|
token: str = jwt.encode(
|
|
payload,
|
|
key=_jwt_secret(),
|
|
algorithm=_jwt_algorithm(),
|
|
)
|
|
return token
|
|
|
|
|
|
def verify_access_token(token: str) -> AccessTokenClaims:
|
|
"""Verify an access JWT and return its claims.
|
|
|
|
Raises TokenInvalidError / TokenExpiredError on failure.
|
|
"""
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
key=_jwt_secret(),
|
|
algorithms=[_jwt_algorithm()],
|
|
audience="contract-check",
|
|
options={
|
|
"require": ["sub", "telegram_id", "exp", "iat"],
|
|
"verify_aud": True,
|
|
},
|
|
)
|
|
except jwt.ExpiredSignatureError as exc:
|
|
raise TokenExpiredError("token expired") from exc
|
|
except jwt.InvalidTokenError as exc:
|
|
raise TokenInvalidError("invalid token") from exc
|
|
|
|
if payload.get("type") != JWT_TYPE_ACCESS:
|
|
raise TokenInvalidError("unexpected token type")
|
|
|
|
try:
|
|
return AccessTokenClaims.from_dict(payload)
|
|
except (KeyError, ValueError, TypeError) as exc:
|
|
raise TokenInvalidError("malformed token claims") from exc
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RefreshTokenClaims:
|
|
"""Payload for a refresh JWT. The `jti` is checked against the refresh store."""
|
|
|
|
sub: uuid.UUID # user_id
|
|
jti: str # opaque id used as the Redis-store key
|
|
type: str
|
|
exp: int | None = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"sub": str(self.sub),
|
|
"jti": self.jti,
|
|
"type": self.type,
|
|
"exp": self.exp,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, payload: dict[str, Any]) -> RefreshTokenClaims:
|
|
return cls(
|
|
sub=uuid.UUID(str(payload["sub"])),
|
|
jti=str(payload["jti"]),
|
|
type=str(payload.get("type", JWT_TYPE_REFRESH)),
|
|
exp=payload.get("exp"),
|
|
)
|
|
|
|
|
|
def create_refresh_token(user_id: uuid.UUID, jti: str) -> str:
|
|
"""Sign a refresh JWT. `jti` is the lookup key in the refresh-token store."""
|
|
now = dt.datetime.now(tz=dt.UTC)
|
|
claims = RefreshTokenClaims(sub=user_id, jti=jti, type=JWT_TYPE_REFRESH)
|
|
payload = claims.to_dict()
|
|
payload.update(
|
|
{
|
|
"iat": int(now.timestamp()),
|
|
"exp": int((now + _jwt_refresh_ttl()).timestamp()),
|
|
"iss": "contract-check",
|
|
"aud": "contract-check",
|
|
}
|
|
)
|
|
token: str = jwt.encode(
|
|
payload,
|
|
key=_jwt_secret(),
|
|
algorithm=_jwt_algorithm(),
|
|
)
|
|
return token
|
|
|
|
|
|
def verify_refresh_token(token: str) -> RefreshTokenClaims:
|
|
"""Verify a refresh JWT signature/expiry. Does NOT check the store — see core.auth_refresh.
|
|
|
|
Raises TokenInvalidError / TokenExpiredError on failure.
|
|
"""
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
key=_jwt_secret(),
|
|
algorithms=[_jwt_algorithm()],
|
|
audience="contract-check",
|
|
options={
|
|
"require": ["sub", "jti", "exp", "iat"],
|
|
"verify_aud": True,
|
|
},
|
|
)
|
|
except jwt.ExpiredSignatureError as exc:
|
|
raise TokenExpiredError("token expired") from exc
|
|
except jwt.InvalidTokenError as exc:
|
|
raise TokenInvalidError("invalid token") from exc
|
|
|
|
if payload.get("type") != JWT_TYPE_REFRESH:
|
|
raise TokenInvalidError("unexpected token type")
|
|
|
|
try:
|
|
return RefreshTokenClaims.from_dict(payload)
|
|
except (KeyError, ValueError, TypeError) as exc:
|
|
raise TokenInvalidError("malformed token claims") from exc
|
|
|
|
|
|
def _telegram_secret_key(bot_token: str) -> bytes:
|
|
"""Telegram uses HMAC_SHA256(BOT_TOKEN, 'WebAppData') as the signing key."""
|
|
return hmac.new(
|
|
bot_token.encode("utf-8"),
|
|
b"WebAppData",
|
|
hashlib.sha256,
|
|
).digest()
|
|
|
|
|
|
def _constant_time_compare(a: str, b: str) -> bool:
|
|
return secrets.compare_digest(a.encode("utf-8"), b.encode("utf-8"))
|
|
|
|
|
|
def verify_telegram_web_payload(payload: dict[str, Any], bot_token: str) -> UserIdentity:
|
|
"""Verify a Telegram Login Widget callback payload.
|
|
|
|
Reference: https://core.telegram.org/widgets/login
|
|
"""
|
|
if not bot_token:
|
|
raise AuthError("telegram_bot_token is not configured")
|
|
|
|
received_hash = payload.get("hash")
|
|
if not isinstance(received_hash, str) or not received_hash:
|
|
raise AuthError("missing hash")
|
|
|
|
auth_date = payload.get("auth_date")
|
|
if not isinstance(auth_date, (int, str)):
|
|
raise AuthError("missing auth_date")
|
|
|
|
try:
|
|
auth_date_int = int(auth_date)
|
|
except ValueError as exc:
|
|
raise AuthError("invalid auth_date") from exc
|
|
|
|
# Reject payloads older than 24 hours to limit replay window.
|
|
now = int(dt.datetime.now(tz=dt.UTC).timestamp())
|
|
if now - auth_date_int > 24 * 60 * 60:
|
|
raise AuthError("telegram auth payload expired")
|
|
|
|
# Build data-check-string from all fields except hash, sorted by key.
|
|
data_check_fields = sorted((k, v) for k, v in payload.items() if k != "hash" and v is not None)
|
|
data_check_string = "\n".join(f"{k}={v}" for k, v in data_check_fields)
|
|
|
|
expected_hash = hmac.new(
|
|
_telegram_secret_key(bot_token),
|
|
data_check_string.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
if not _constant_time_compare(received_hash, expected_hash):
|
|
raise AuthError("telegram signature mismatch")
|
|
|
|
telegram_id = payload.get("id")
|
|
if not isinstance(telegram_id, int) or telegram_id <= 0:
|
|
raise AuthError("missing telegram id")
|
|
|
|
return UserIdentity(telegram_id=telegram_id)
|
|
|
|
|
|
def verify_telegram_miniapp_init_data(init_data: str, bot_token: str) -> UserIdentity:
|
|
"""Verify Telegram Mini App `initData` and extract the user identity.
|
|
|
|
`initData` is a query-string-like string that contains a `hash` parameter
|
|
and (optionally) a JSON `user` parameter. Telegram signs the full string
|
|
excluding `hash` using HMAC_SHA256(bot_token, "WebAppData").
|
|
|
|
Reference: https://core.telegram.org/bots/webapps#validating-data-received-via-the-mini-app
|
|
"""
|
|
if not bot_token:
|
|
raise AuthError("telegram_bot_token is not configured")
|
|
|
|
if not isinstance(init_data, str) or "=" not in init_data:
|
|
raise AuthError("invalid init_data")
|
|
|
|
params = dict(parse_qsl(init_data, keep_blank_values=True))
|
|
received_hash = params.pop("hash", None)
|
|
if not received_hash:
|
|
raise AuthError("missing hash")
|
|
|
|
data_check_string = "\n".join(f"{k}={v}" for k, v in sorted(params.items()))
|
|
expected_hash = hmac.new(
|
|
_telegram_secret_key(bot_token),
|
|
data_check_string.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
if not _constant_time_compare(received_hash, expected_hash):
|
|
raise AuthError("telegram signature mismatch")
|
|
|
|
user_json = params.get("user")
|
|
if not user_json:
|
|
raise AuthError("missing user in init_data")
|
|
|
|
import json
|
|
|
|
try:
|
|
user = json.loads(user_json)
|
|
except json.JSONDecodeError as exc:
|
|
raise AuthError("invalid user json") from exc
|
|
|
|
telegram_id = user.get("id")
|
|
if not isinstance(telegram_id, int) or telegram_id <= 0:
|
|
raise AuthError("missing telegram id")
|
|
|
|
return UserIdentity(telegram_id=telegram_id)
|
|
|
|
|
|
def verify_bot_identity(telegram_id: int) -> UserIdentity:
|
|
"""Identity proof used by the trusted bot adapter.
|
|
|
|
The bot receives `message.from_user.id` directly from Telegram; here we just
|
|
validate it is a positive integer. The API caller (the bot) is authenticated
|
|
separately via its service token.
|
|
"""
|
|
if not isinstance(telegram_id, int) or telegram_id <= 0:
|
|
raise AuthError("invalid telegram_id")
|
|
return UserIdentity(telegram_id=telegram_id)
|