ElebobaandClaude Opus 5.5
9362eb6ecc
CI / ruff · pytest (pull_request) Successful in 1m26s
- setup() at package import (idempotent, replaces only base.py defaults): MisMock, ConnectCodeIdentity, GrlsExcerpts, CompositeNotifier; outbox stays MemoryOutbox - MisMock: clinical DB prescriptions, fallback data/meds/mis_prescriptions.json (synthetic, by token) - ConnectCodeIdentity: one-time connect code via identity_repo, 15 min grace for the double check in one connect flow (Telegram /start + engine `new` state) - GrlsExcerpts: synthetic paraphrased instruction excerpts for sertraline, paroxetine, escitalopram, fluoxetine, mirtazapine, prazosin - CompositeNotifier: in-memory feed (recent(since_id)) + Telegram doctor chat, PII-free, never raises - test_llm_med_qa: default-registry test now patches NoDrugInfo explicitly Refs #14 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
59 lines
2.5 KiB
Python
59 lines
2.5 KiB
Python
"""ЕСИА stand-in: the patient proves identity with the one-time connect code issued by the doctor.
|
|
|
|
The doctor creates the code in «Подключить пациента» (identity DB `connect_codes`); the patient opens the
|
|
QR deep link / enters the code. In prod this is ЕСИА OAuth; the `IdentityProvider` Protocol stays the same.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import UTC, timedelta
|
|
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from app.core import clock
|
|
from app.integrations.base import VerifiedPerson
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# One connect flow verifies the same code twice: the Telegram channel on `/start <code>` (to bind the chat)
|
|
# and then the dialog engine's `new` state (to check the code belongs to this token). A short grace window
|
|
# lets the second check pass; after it a used code is rejected like an unknown one.
|
|
REDEEM_GRACE = timedelta(minutes=15)
|
|
|
|
|
|
class ConnectCodeIdentity:
|
|
"""`IdentityProvider` over identity-DB connect codes."""
|
|
|
|
def __init__(self, grace: timedelta = REDEEM_GRACE) -> None:
|
|
self.grace = grace
|
|
|
|
async def verify(self, auth_code: str) -> VerifiedPerson | None:
|
|
code = auth_code.strip()
|
|
if not code:
|
|
return None
|
|
# Imported lazily: the integrations package is imported by clinical-side modules, which must not
|
|
# pull the identity schema into their import graph (guardrail #5).
|
|
from app.db import identity_repo
|
|
from app.db.identity import ConnectCode
|
|
from app.db.session import get_identity_session
|
|
|
|
now = clock.now()
|
|
try:
|
|
async with get_identity_session() as s:
|
|
row = await s.get(ConnectCode, code)
|
|
if row is None:
|
|
log.info("connect code rejected: unknown")
|
|
return None
|
|
person = VerifiedPerson(token=row.token, doctor_id=row.doctor_id)
|
|
if row.used_at is not None:
|
|
used_at = row.used_at if row.used_at.tzinfo else row.used_at.replace(tzinfo=UTC)
|
|
if now - used_at <= self.grace:
|
|
return person
|
|
log.info("connect code rejected: already used")
|
|
return None
|
|
if await identity_repo.redeem_connect_code(s, code=code, now=now) is None:
|
|
return None
|
|
return person
|
|
except (OSError, SQLAlchemyError) as exc:
|
|
log.warning("connect code check failed error=%s", type(exc).__name__)
|
|
return None
|