diff --git a/codebase/backend/.env.example b/codebase/backend/.env.example index 59c518d..77ae157 100644 --- a/codebase/backend/.env.example +++ b/codebase/backend/.env.example @@ -139,6 +139,7 @@ BITRIX_PUBLIC_BASE_URL=https://chat.example.ru/bitrix # Generate with: python -c "import base64,secrets; print(base64.urlsafe_b64encode(secrets.token_bytes(32)).decode())" BITRIX_TOKEN_ENCRYPTION_KEY=change-me BITRIX_HTTP_TIMEOUT_SEC=10 +BITRIX_LOCAL_APP_HTTP_TIMEOUT_SEC=20 SELECTEL_S3_ENDPOINT_URL=https://s3.ru-7.storage.selcloud.ru SELECTEL_S3_BUCKET_DOCUMENTS=han-chat-documents diff --git a/codebase/backend/api-backend/alembic/versions/0003_consent_device_audit_context.py b/codebase/backend/api-backend/alembic/versions/0003_consent_device_audit_context.py new file mode 100644 index 0000000..e3d6445 --- /dev/null +++ b/codebase/backend/api-backend/alembic/versions/0003_consent_device_audit_context.py @@ -0,0 +1,33 @@ +"""Store consent device snapshots and remove audit IP. + +Revision ID: 0003_consent_audit +Revises: 0002_pgcrypto_digest +Create Date: 2026-07-16 +""" + +from collections.abc import Sequence + +from alembic import op +revision: str = "0003_consent_audit" +down_revision: str | None = "0002_pgcrypto_digest" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute( + """ + ALTER TABLE han_app.user_consents + ADD COLUMN IF NOT EXISTS device_json jsonb NOT NULL DEFAULT '{}'::jsonb + """ + ) + op.execute( + """ + ALTER TABLE han_app.audit_events + DROP COLUMN IF EXISTS ip + """ + ) + + +def downgrade() -> None: + raise RuntimeError("Consent and audit context migration is forward-only") diff --git a/codebase/backend/api-backend/app/db.py b/codebase/backend/api-backend/app/db.py index ebe9c07..4dc0451 100644 --- a/codebase/backend/api-backend/app/db.py +++ b/codebase/backend/api-backend/app/db.py @@ -17,7 +17,7 @@ from sqlalchemy import ( UniqueConstraint, func, ) -from sqlalchemy.dialects.postgresql import INET, UUID +from sqlalchemy.dialects.postgresql import INET, JSONB, UUID from sqlalchemy.ext.asyncio import ( AsyncEngine, AsyncSession, @@ -86,6 +86,7 @@ class UserConsent(Common, Base): accepted_at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) client_ip: Mapped[str | None] = mapped_column(INET) user_agent_hash: Mapped[str | None] = mapped_column(String(64)) + device_json: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict) class UxSession(Common, Base): @@ -275,7 +276,6 @@ class AuditEvent(Base): trace_id: Mapped[str | None] = mapped_column(String(64)) resource_type: Mapped[str | None] = mapped_column(String(64)) resource_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True)) - ip: Mapped[str | None] = mapped_column(INET) user_agent_hash: Mapped[str | None] = mapped_column(String(64)) outcome: Mapped[str] = mapped_column(String(32)) metadata_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict) diff --git a/codebase/backend/api-backend/app/main.py b/codebase/backend/api-backend/app/main.py index 25dcb50..79ea173 100644 --- a/codebase/backend/api-backend/app/main.py +++ b/codebase/backend/api-backend/app/main.py @@ -40,6 +40,7 @@ from app.db import ( PopularQuestion, TextResource, UserIdentity, + UxSession, ) from app.integrations import ( DependencyFailure, @@ -62,6 +63,7 @@ from app.schemas import ( encode_cursor, ) from app.services import ( + AuditContext, DomainError, SettingsSnapshot, apply_inbox, @@ -153,7 +155,7 @@ app = FastAPI( log = structlog.get_logger() -def client_ip(request: Request) -> str: +def client_ip(request: Request) -> str | None: """Trust forwarded client addresses only from configured reverse proxies.""" peer = request.client.host if request.client else "" try: @@ -170,7 +172,40 @@ def client_ip(request: Request) -> str: try: return str(ip_address(candidate)) except ValueError: - return "unknown" + return None + + +def request_trace_id(request: Request) -> str: + parts = request.headers.get("traceparent", "").lower().split("-") + if ( + len(parts) == 4 + and len(parts[0]) == 2 + and len(parts[1]) == 32 + and len(parts[2]) == 16 + and len(parts[3]) == 2 + ): + try: + int("".join(parts), 16) + if parts[1] != "0" * 32 and parts[2] != "0" * 16: + return parts[1] + except ValueError: + pass + return uuid.uuid4().hex + + +def user_agent_hash(request: Request) -> str | None: + value = " ".join(request.headers.get("User-Agent", "").split()) + return hashlib.sha256(value[:1024].encode()).hexdigest() if value else None + + +def audit_context(request: Request, ux_session_id: uuid.UUID | None = None) -> AuditContext: + return AuditContext( + request_id=request.state.request_id, + trace_id=request.state.trace_id, + ux_session_id=ux_session_id, + user_agent_hash=request.state.user_agent_hash, + client_ip=client_ip(request), + ) @app.middleware("http") @@ -181,10 +216,13 @@ async def request_context(request: Request, call_next: Any) -> Response: except ValueError: request_id = str(uuid.uuid4()) request.state.request_id = request_id + request.state.trace_id = request_trace_id(request) + request.state.user_agent_hash = user_agent_hash(request) request.state.started_at = time.monotonic() structlog.contextvars.clear_contextvars() structlog.contextvars.bind_contextvars( request_id=request_id, + trace_id=request.state.trace_id, ux_session_id=request.headers.get("X-Ux-Session-Id"), method=request.method, route=request.url.path, @@ -291,6 +329,26 @@ async def current_user(db: Session, auth: PrincipalDep) -> UserIdentity: UserDep = Annotated[UserIdentity, Depends(current_user)] +async def required_user_audit_context( + request: Request, + db: Session, + user: UserDep, + x_ux_session_id: Annotated[str, Header(alias="X-Ux-Session-Id")], +) -> AuditContext: + session_id = ux_id(x_ux_session_id) + if session_id is None: + raise DomainError("validation_error", 400, "X-Ux-Session-Id is required") + exists = await db.scalar( + select(UxSession.id).where(UxSession.id == session_id, UxSession.user_id == user.id) + ) + if exists is None: + raise DomainError("validation_error", 400, "X-Ux-Session-Id is invalid") + return audit_context(request, session_id) + + +UserAuditContextDep = Annotated[AuditContext, Depends(required_user_audit_context)] + + async def snapshot(db: Session) -> SettingsSnapshot: return await load_settings(db) @@ -357,7 +415,7 @@ async def ready(request: Request, db: Session): try: await db.execute(text("SELECT 1")) revision = await db.scalar(text("SELECT version_num FROM han_app.alembic_version LIMIT 1")) - if revision != "0002_pgcrypto_digest": + if revision != "0003_consent_audit": raise RuntimeError("unexpected database revision") await load_settings(db) components["postgres"] = "ok" @@ -400,7 +458,7 @@ async def app_config(request: Request, response: Response, settings: SnapshotDep await enforce_limit( request, "ip", - client_ip(request), + client_ip(request) or "unknown", "public", settings.limit("rate_limit.public_endpoints.per_ip"), fail_closed=False, @@ -444,7 +502,7 @@ async def content( await enforce_limit( request, "ip", - client_ip(request), + client_ip(request) or "unknown", "public", settings.limit("rate_limit.public_endpoints.per_ip"), fail_closed=False, @@ -480,7 +538,7 @@ async def content( async def auth_bootstrap( body: BootstrapRequest, request: Request, db: Session, auth: PrincipalDep, settings: SnapshotDep ): - return await bootstrap(db, auth, body, settings, request.state.request_id) + return await bootstrap(db, auth, body, settings, audit_context(request)) @app.post("/api/v1/consents", status_code=201, tags=["auth"]) @@ -490,18 +548,16 @@ async def consents( db: Session, user: UserDep, settings: SnapshotDep, - x_ux_session_id: Annotated[str | None, Header()] = None, + context: UserAuditContextDep, ): - return await record_consents( - db, user, body, settings, request.state.request_id, ux_id(x_ux_session_id) - ) + return await record_consents(db, user, body, settings, context) @app.post("/api/v1/analytics/session-start", status_code=201, tags=["analytics"]) async def analytics_session( body: SessionStartRequest, request: Request, db: Session, user: UserDep ): - return await start_session(db, user, body, request.state.request_id) + return await start_session(db, user, body, audit_context(request)) @app.get("/api/v1/me", tags=["profile"]) @@ -564,7 +620,7 @@ async def document_download( db: Session, user: UserDep, settings: SnapshotDep, - x_ux_session_id: Annotated[str | None, Header()] = None, + context: UserAuditContextDep, ): await enforce_limit( request, @@ -578,11 +634,11 @@ async def document_download( db.add( audit( "document.download_url_issued", - request.state.request_id, + context, user.id, "document", item.id, - ux_id(x_ux_session_id), + metadata={"expires_in_seconds": 300}, ) ) await db.commit() @@ -595,6 +651,7 @@ async def dialogs_create( request: Request, db: Session, user: UserDep, + context: UserAuditContextDep, idempotency_key: Annotated[str | None, Header()] = None, ): key = await required_idempotency(idempotency_key) @@ -608,7 +665,7 @@ async def dialogs_create( if cached["fingerprint"] != fingerprint: raise DomainError("idempotency_key_reused", 409, "Idempotency key was reused") return JSONResponse(cached["body"], status_code=cached["status"]) - body, status = await create_dialog(db, user, request.state.request_id, key) + body, status = await create_dialog(db, user, context, key) try: await request.app.state.idempotency.put( scope, @@ -735,6 +792,7 @@ async def message_create( db: Session, user: UserDep, business: SnapshotDep, + context: UserAuditContextDep, idempotency_key: Annotated[str | None, Header()] = None, ): key = await required_idempotency(idempotency_key) @@ -772,7 +830,7 @@ async def message_create( dialog_id, body, key, - request.state.request_id, + context, request.app.state.settings, request.app.state.safety, request.app.state.openlines, @@ -803,6 +861,7 @@ async def attachment_init( db: Session, user: UserDep, settings: SnapshotDep, + context: UserAuditContextDep, ): await enforce_limit( request, @@ -812,9 +871,7 @@ async def attachment_init( settings.limit("rate_limit.message_send.per_user"), fail_closed=True, ) - return await init_attachment( - db, user, dialog_id, body, settings, request.app.state.s3, request.state.request_id - ) + return await init_attachment(db, user, dialog_id, body, settings, request.app.state.s3, context) @app.post( @@ -828,9 +885,10 @@ async def attachment_complete( request: Request, db: Session, user: UserDep, + context: UserAuditContextDep, ): return await complete_attachment( - db, user, dialog_id, attachment_id, body, request.app.state.s3, request.state.request_id + db, user, dialog_id, attachment_id, body, request.app.state.s3, context ) @@ -845,7 +903,7 @@ async def attachment_download( db: Session, user: UserDep, settings: SnapshotDep, - x_ux_session_id: Annotated[str | None, Header()] = None, + context: UserAuditContextDep, ): await enforce_limit( request, @@ -871,11 +929,11 @@ async def attachment_download( db.add( audit( "attachment.download_url_issued", - request.state.request_id, + context, user.id, "attachment", item.id, - ux_id(x_ux_session_id), + metadata={"expires_in_seconds": 300}, ) ) await db.commit() @@ -895,7 +953,7 @@ async def inbox(event: OpenLinesInbox, request: Request, db: Session, settings: body, status = await apply_inbox( db, event, - request.state.request_id, + audit_context(request), settings, request.app.state.s3, request.app.state.http, diff --git a/codebase/backend/api-backend/app/services.py b/codebase/backend/api-backend/app/services.py index bcab730..39d538e 100644 --- a/codebase/backend/api-backend/app/services.py +++ b/codebase/backend/api-backend/app/services.py @@ -9,7 +9,7 @@ from pathlib import PurePath from typing import Any import httpx -from sqlalchemy import func, select +from sqlalchemy import case, func, select from sqlalchemy.dialects.postgresql import insert from sqlalchemy.ext.asyncio import AsyncSession @@ -106,6 +106,24 @@ class SettingsSnapshot: return int(amount), windows[period] +@dataclass(frozen=True, slots=True) +class AuditContext: + request_id: str + trace_id: str + ux_session_id: uuid.UUID | None + user_agent_hash: str | None + client_ip: str | None + + def with_ux_session(self, ux_session_id: uuid.UUID) -> "AuditContext": + return AuditContext( + request_id=self.request_id, + trace_id=self.trace_id, + ux_session_id=ux_session_id, + user_agent_hash=self.user_agent_hash, + client_ip=self.client_ip, + ) + + async def load_settings(session: AsyncSession) -> SettingsSnapshot: rows = ( await session.execute(select(AppSetting).where(AppSetting.record_status == "A")) @@ -138,26 +156,45 @@ async def resolve_user(session: AsyncSession, principal: Principal) -> UserIdent def audit( event_type: str, - request_id: str, + context: AuditContext, user_id: uuid.UUID | None, resource_type: str | None = None, resource_id: uuid.UUID | None = None, - ux_session_id: uuid.UUID | None = None, metadata: dict[str, Any] | None = None, + outcome: str = "success", ) -> AuditEvent: return AuditEvent( event_type=event_type, actor_type="user" if user_id else "service", user_id=user_id, - ux_session_id=ux_session_id, - request_id=request_id, + ux_session_id=context.ux_session_id, + request_id=context.request_id, + trace_id=context.trace_id, resource_type=resource_type, resource_id=resource_id, - outcome="success", + user_agent_hash=context.user_agent_hash, + outcome=outcome, metadata_json=metadata or {}, ) +def device_snapshot(device: Any) -> dict[str, str | None]: + return { + "platform": device.platform, + "app_version": device.app_version, + "device_id_hash": ( + hashlib.sha256(device.device_id.encode()).hexdigest() if device.device_id else None + ), + } + + +def consent_versions(consents: Any) -> dict[str, str]: + return { + name: getattr(consents, name).version + for name in ("personal_data", "user_agreement", "marketing") + } + + def validate_consents(consents: Any, snapshot: SettingsSnapshot) -> None: for name in ("personal_data", "user_agreement", "marketing"): choice = getattr(consents, name) @@ -174,12 +211,13 @@ async def bootstrap( principal: Principal, body: BootstrapRequest, snapshot: SettingsSnapshot, - request_id: str, + context: AuditContext, ) -> dict[str, Any]: if not principal.phone_number: raise DomainError("phone_claim_missing", 400, "Verified phone claim is missing") validate_consents(body.consents, snapshot) now = datetime.now(UTC) + device = device_snapshot(body.device) statement = ( insert(UserIdentity) .values( @@ -211,6 +249,9 @@ async def bootstrap( document_version=choice.version, accepted=choice.accepted, accepted_at=now, + client_ip=context.client_ip, + user_agent_hash=context.user_agent_hash, + device_json=device, ) .on_conflict_do_nothing( index_elements=[ @@ -220,7 +261,18 @@ async def bootstrap( ] ) ) - session.add(audit("auth.bootstrap", request_id, user_id)) + session.add( + audit( + "auth.bootstrap", + context, + user_id, + metadata={ + "consent_versions": consent_versions(body.consents), + "platform": body.device.platform, + "app_version": body.device.app_version, + }, + ) + ) await session.commit() return {"user_id": user_id, "profile_ready": True} @@ -230,42 +282,83 @@ async def record_consents( user: UserIdentity, body: ConsentsRequest, snapshot: SettingsSnapshot, - request_id: str, - ux_session_id: uuid.UUID | None, + context: AuditContext, ) -> dict[str, Any]: validate_consents(body.consents, snapshot) + if context.ux_session_id is None: + raise DomainError("validation_error", 400, "X-Ux-Session-Id is required") + ux_session = ( + await session.execute( + select(UxSession).where( + UxSession.id == context.ux_session_id, + UxSession.user_id == user.id, + ) + ) + ).scalar_one_or_none() + if ux_session is None: + raise DomainError("validation_error", 400, "X-Ux-Session-Id is invalid") + device = { + "platform": ux_session.platform, + "app_version": ux_session.app_version, + "device_id_hash": ux_session.device_id_hash, + } now = datetime.now(UTC) versions: dict[str, str] = {} for consent_type in ("personal_data", "user_agreement", "marketing"): choice = getattr(body.consents, consent_type) versions[consent_type] = choice.version - await session.execute( - insert(UserConsent) - .values( - id=uuid.uuid4(), - user_id=user.id, - ux_session_id=ux_session_id, - consent_type=consent_type, - document_version=choice.version, - accepted=choice.accepted, - accepted_at=now, - ) - .on_conflict_do_nothing() + statement = insert(UserConsent).values( + id=uuid.uuid4(), + user_id=user.id, + ux_session_id=context.ux_session_id, + consent_type=consent_type, + document_version=choice.version, + accepted=choice.accepted, + accepted_at=now, + client_ip=context.client_ip, + user_agent_hash=context.user_agent_hash, + device_json=device, ) - session.add(audit("consent.recorded", request_id, user.id, ux_session_id=ux_session_id)) + await session.execute( + statement.on_conflict_do_update( + index_elements=[ + UserConsent.user_id, + UserConsent.consent_type, + UserConsent.document_version, + ], + set_={ + "ux_session_id": statement.excluded.ux_session_id, + "client_ip": func.coalesce( + UserConsent.client_ip, statement.excluded.client_ip + ), + "user_agent_hash": func.coalesce( + UserConsent.user_agent_hash, + statement.excluded.user_agent_hash, + ), + "device_json": case( + (UserConsent.device_json == {}, statement.excluded.device_json), + else_=UserConsent.device_json, + ), + }, + ) + ) + session.add( + audit( + "consent.recorded", + context, + user.id, + metadata={"consent_versions": versions}, + ) + ) await session.commit() return {"recorded_at": now, "versions": versions} async def start_session( - session: AsyncSession, user: UserIdentity, body: SessionStartRequest, request_id: str + session: AsyncSession, user: UserIdentity, body: SessionStartRequest, context: AuditContext ) -> dict[str, Any]: now, session_id = datetime.now(UTC), uuid.uuid4() - device_hash = ( - hashlib.sha256(body.device.device_id.encode()).hexdigest() - if body.device.device_id - else None - ) + device = device_snapshot(body.device) session.add( UxSession( id=session_id, @@ -273,11 +366,22 @@ async def start_session( start_reason=body.start_reason, platform=body.device.platform, app_version=body.device.app_version, - device_id_hash=device_hash, + device_id_hash=device["device_id_hash"], started_at=now, ) ) - session.add(audit("session_start", request_id, user.id, ux_session_id=session_id)) + session.add( + audit( + "session_start", + context.with_ux_session(session_id), + user.id, + metadata={ + "start_reason": body.start_reason, + "platform": body.device.platform, + "app_version": body.device.app_version, + }, + ) + ) await session.commit() return {"ux_session_id": session_id, "started_at": now} @@ -407,7 +511,7 @@ async def owned_dialog(session: AsyncSession, user_id: uuid.UUID, dialog_id: uui async def create_dialog( - session: AsyncSession, user: UserIdentity, request_id: str, idempotency_key: str + session: AsyncSession, user: UserIdentity, context: AuditContext, idempotency_key: str ) -> tuple[dict[str, Any], int]: scope = "dialogs.create" fingerprint = hashlib.sha256(f"POST:/api/v1/dialogs:{user.id}".encode()).hexdigest() @@ -454,7 +558,16 @@ async def create_dialog( return result, 200 dialog = Dialog(id=uuid.uuid4(), user_id=user.id, status="open") session.add(dialog) - session.add(audit("dialog.created", request_id, user.id, "dialog", dialog.id)) + session.add( + audit( + "dialog.created", + context, + user.id, + "dialog", + dialog.id, + metadata={"status": dialog.status}, + ) + ) result = dialog_dto(dialog) session.add( IdempotencyRecord( @@ -482,7 +595,7 @@ async def init_attachment( body: AttachmentInitRequest, snapshot: SettingsSnapshot, s3: S3Client, - request_id: str, + context: AuditContext, ) -> dict[str, Any]: await owned_dialog(session, user.id, dialog_id) extension = PurePath(body.file_name).suffix.lower().lstrip(".") @@ -514,7 +627,14 @@ async def init_attachment( ) session.add(item) session.add( - audit("attachment.upload_initialized", request_id, user.id, "attachment", attachment_id) + audit( + "attachment.upload_initialized", + context, + user.id, + "attachment", + attachment_id, + metadata={"mime_type": body.mime_type, "size_bytes": body.size_bytes}, + ) ) await session.commit() url = await s3.presign_put(key, body.mime_type, ttl) @@ -533,7 +653,7 @@ async def complete_attachment( attachment_id: uuid.UUID, body: AttachmentCompleteRequest, s3: S3Client, - request_id: str, + context: AuditContext, ) -> dict[str, Any]: item = ( await session.execute( @@ -560,7 +680,16 @@ async def complete_attachment( raise DomainError("attachment_checksum_mismatch", 400, "Uploaded metadata does not match") item.checksum_sha256 = checksum item.completed_at = datetime.now(UTC) - session.add(audit("attachment.upload_completed", request_id, user.id, "attachment", item.id)) + session.add( + audit( + "attachment.upload_completed", + context, + user.id, + "attachment", + item.id, + metadata={"mime_type": item.mime_type, "size_bytes": item.size_bytes}, + ) + ) await session.commit() return attachment_dto(item) @@ -583,7 +712,7 @@ async def send_message( dialog_id: uuid.UUID, body: MessageRequest, idem_key: str, - request_id: str, + context: AuditContext, settings: Settings, safety: SafetyClient, openlines: OpenLinesClient, @@ -672,7 +801,16 @@ async def send_message( session.add(message) if attachment: attachment.message_id = message_id - session.add(audit("message.submitted", request_id, user.id, "message", message_id)) + session.add( + audit( + "message.submitted", + context, + user.id, + "message", + message_id, + metadata={"content_kind": body.content_kind, "delivery_status": "processing"}, + ) + ) await session.commit() await publish_message(fanout, message, settings, [attachment] if attachment else []) payload: dict[str, Any] = {"message_id": str(message_id), "content_kind": kind, "text": text} @@ -684,8 +822,9 @@ async def send_message( "mime_type": attachment.mime_type, "size_bytes": attachment.size_bytes, } + outbox: DeliveryOutbox | None = None try: - verdict = await safety.check(payload, request_id) + verdict = await safety.check(payload, context.request_id) if verdict["_status"] == 203: task_id = verdict["task_id"] task = SafetyTask( @@ -703,7 +842,7 @@ async def send_message( deadline = time_monotonic() + settings.message_safety_task_poll_max_sec while time_monotonic() < deadline: await sleep(settings.message_safety_task_poll_interval_sec) - verdict = await safety.poll(task_id, request_id) + verdict = await safety.poll(task_id, context.request_id) if verdict["_status"] != 203: break else: @@ -719,7 +858,20 @@ async def send_message( if attachment and attachment.quarantine_object_key: attachment.scan_status = "infected" await s3.delete_quarantine(attachment.quarantine_object_key) - session.add(audit("message.blocked", request_id, user.id, "message", message.id)) + session.add( + audit( + "message.blocked", + context, + user.id, + "message", + message.id, + metadata={ + "content_kind": message.content_kind, + "safety_status": message.safety_status, + "delivery_status": message.delivery_status, + }, + ) + ) idem.status = "completed" idem.response_status = 422 idem.response_body_json = { @@ -765,18 +917,35 @@ async def send_message( ), }, }, - next_attempt_at=datetime.now(UTC), + # Keep the row recoverable after a process crash, but do not let the + # delivery worker race the synchronous first attempt. + next_attempt_at=datetime.now(UTC) + + timedelta(seconds=settings.bitrix_local_app_http_timeout_sec + 5), ) session.add(outbox) await session.commit() await publish_message_status(fanout, message, settings) delivery_payload = await fresh_openlines_payload(outbox.payload_json, s3) - await openlines.send(message.id, delivery_payload, request_id) + await openlines.send(message.id, delivery_payload, context.request_id) message.delivery_status = "delivered" dialog.status = "waiting_for_company" dialog.last_message_at = datetime.now(UTC) outbox.status = "delivered" - session.add(audit("message.delivered", request_id, user.id, "message", message.id)) + session.add( + audit( + "message.delivered", + context, + user.id, + "message", + message.id, + metadata={ + "content_kind": message.content_kind, + "safety_status": message.safety_status, + "delivery_status": message.delivery_status, + }, + outcome="blocked", + ) + ) result = message_dto(message, [attachment] if attachment else []) idem.status = "completed" idem.response_status = 201 @@ -789,7 +958,31 @@ async def send_message( raise except DependencyFailure as exc: message.delivery_status = "failed" - session.add(audit("message.failed", request_id, user.id, "message", message.id)) + if outbox is not None: + outbox.attempt_count += 1 + outbox.status = "retry" + outbox.next_attempt_at = datetime.now(UTC) + timedelta( + seconds=min(3600, 2**outbox.attempt_count) + ) + outbox.last_error_code = ( + "dependency_timeout" if exc.timeout else "dependency_unavailable" + ) + outbox.locked_at = None + outbox.locked_by = None + session.add( + audit( + "message.failed", + context, + user.id, + "message", + message.id, + metadata={ + "content_kind": message.content_kind, + "delivery_status": message.delivery_status, + }, + outcome="failed", + ) + ) await session.commit() await publish_message_status(fanout, message, settings) raise DomainError( @@ -802,7 +995,7 @@ async def send_message( async def apply_inbox( session: AsyncSession, event: OpenLinesInbox, - request_id: str, + context: AuditContext, snapshot: SettingsSnapshot, s3: S3Client, http: httpx.AsyncClient, @@ -910,7 +1103,16 @@ async def apply_inbox( dialog.status = "waiting_for_client" dialog.last_message_at = event.occurred_at receipt.status = "applied" - session.add(audit("openlines.inbox_applied", request_id, dialog.user_id, "dialog", dialog.id)) + session.add( + audit( + "openlines.inbox_applied", + context, + None, + "dialog", + dialog.id, + metadata={"event_type": event.event_type}, + ) + ) await session.commit() if message is not None: await publish_message(fanout, message, settings, [attachment] if attachment else []) diff --git a/codebase/backend/api-backend/app/settings.py b/codebase/backend/api-backend/app/settings.py index fdf5b4b..995e263 100644 --- a/codebase/backend/api-backend/app/settings.py +++ b/codebase/backend/api-backend/app/settings.py @@ -45,7 +45,7 @@ class Settings(BaseSettings): bitrix_local_app_internal_token: SecretStr = Field(alias="BITRIX_LOCAL_APP_INTERNAL_TOKEN") bitrix_api_inbox_token: SecretStr = Field(alias="BITRIX_API_INBOX_TOKEN") bitrix_local_app_http_timeout_sec: float = Field( - default=10, alias="BITRIX_LOCAL_APP_HTTP_TIMEOUT_SEC" + default=20, alias="BITRIX_LOCAL_APP_HTTP_TIMEOUT_SEC" ) bitrix_local_app_circuit_failure_threshold: int = Field( default=5, alias="BITRIX_LOCAL_APP_CIRCUIT_FAILURE_THRESHOLD" diff --git a/codebase/backend/api-backend/docker-compose.yml b/codebase/backend/api-backend/docker-compose.yml index 059d5bc..6782a88 100644 --- a/codebase/backend/api-backend/docker-compose.yml +++ b/codebase/backend/api-backend/docker-compose.yml @@ -25,7 +25,7 @@ services: BITRIX_LOCAL_APP_BASE_URL: ${BITRIX_LOCAL_APP_BASE_URL} BITRIX_LOCAL_APP_INTERNAL_TOKEN: ${BITRIX_LOCAL_APP_INTERNAL_TOKEN} BITRIX_API_INBOX_TOKEN: ${BITRIX_API_INBOX_TOKEN} - BITRIX_LOCAL_APP_HTTP_TIMEOUT_SEC: ${BITRIX_LOCAL_APP_HTTP_TIMEOUT_SEC:-10} + BITRIX_LOCAL_APP_HTTP_TIMEOUT_SEC: ${BITRIX_LOCAL_APP_HTTP_TIMEOUT_SEC:-20} BITRIX_LOCAL_APP_CIRCUIT_FAILURE_THRESHOLD: ${BITRIX_LOCAL_APP_CIRCUIT_FAILURE_THRESHOLD:-5} BITRIX_LOCAL_APP_CIRCUIT_OPEN_SEC: ${BITRIX_LOCAL_APP_CIRCUIT_OPEN_SEC:-30} KEYCLOAK_SETTINGS_BRIDGE_TOKEN: ${KEYCLOAK_SETTINGS_BRIDGE_TOKEN} diff --git a/codebase/backend/api-backend/tests/unit/test_audit_context.py b/codebase/backend/api-backend/tests/unit/test_audit_context.py new file mode 100644 index 0000000..d487227 --- /dev/null +++ b/codebase/backend/api-backend/tests/unit/test_audit_context.py @@ -0,0 +1,154 @@ +import hashlib +import uuid +from types import SimpleNamespace + +import pytest +from starlette.requests import Request + +from app.main import ( + app, + client_ip, + request_trace_id, + required_user_audit_context, + user_agent_hash, +) +from app.schemas import Device +from app.services import AuditContext, DomainError, audit, device_snapshot + + +def request( + *, + peer: str = "172.18.0.5", + forwarded: str | None = None, + traceparent: str | None = None, + user_agent: str | None = None, + ux_session_id: str | None = None, + trusted: str = "172.16.0.0/12", +) -> Request: + headers: list[tuple[bytes, bytes]] = [] + for name, value in ( + ("x-forwarded-for", forwarded), + ("traceparent", traceparent), + ("user-agent", user_agent), + ("x-ux-session-id", ux_session_id), + ): + if value is not None: + headers.append((name.encode(), value.encode())) + settings = SimpleNamespace(trusted_proxy_cidrs=trusted) + app = SimpleNamespace(state=SimpleNamespace(settings=settings)) + return Request( + { + "type": "http", + "method": "GET", + "path": "/", + "headers": headers, + "client": (peer, 12345), + "server": ("test", 443), + "scheme": "https", + "query_string": b"", + "app": app, + } + ) + + +def test_client_ip_only_trusts_forwarded_header_from_configured_proxy() -> None: + assert client_ip(request(forwarded="203.0.113.10")) == "203.0.113.10" + assert ( + client_ip(request(peer="198.51.100.7", forwarded="203.0.113.10")) + == "198.51.100.7" + ) + assert client_ip(request(peer="not-an-ip")) is None + + +def test_trace_id_uses_valid_w3c_header_and_generates_fallback() -> None: + trace_id = "1" * 32 + assert request_trace_id(request(traceparent=f"00-{trace_id}-{'2' * 16}-01")) == trace_id + fallback = request_trace_id(request(traceparent="invalid")) + assert len(fallback) == 32 + int(fallback, 16) + + +def test_user_agent_and_device_are_hashed_without_raw_identifiers() -> None: + agent = "Example Browser/1.0" + assert user_agent_hash(request(user_agent=agent)) == hashlib.sha256(agent.encode()).hexdigest() + snapshot = device_snapshot( + Device(platform="web", app_version="1.2.3", device_id="raw-device-id") + ) + assert snapshot == { + "platform": "web", + "app_version": "1.2.3", + "device_id_hash": hashlib.sha256(b"raw-device-id").hexdigest(), + } + assert "raw-device-id" not in str(snapshot) + + +def test_audit_copies_request_context_and_bounded_metadata() -> None: + session_id = uuid.uuid4() + user_id = uuid.uuid4() + context = AuditContext( + request_id=str(uuid.uuid4()), + trace_id="a" * 32, + ux_session_id=session_id, + user_agent_hash="b" * 64, + client_ip="203.0.113.10", + ) + + event = audit( + "dialog.created", + context, + user_id, + "dialog", + uuid.uuid4(), + metadata={"status": "open"}, + ) + + assert event.user_id == user_id + assert event.ux_session_id == session_id + assert event.trace_id == "a" * 32 + assert event.user_agent_hash == "b" * 64 + assert event.metadata_json == {"status": "open"} + assert event.outcome == "success" + assert not hasattr(event, "ip") + + failed = audit("message.failed", context, user_id, outcome="failed") + assert failed.outcome == "failed" + + +def test_post_session_routes_publish_required_ux_header_in_openapi() -> None: + operation = app.openapi()["paths"]["/api/v1/dialogs"]["post"] + ux_header = next( + parameter + for parameter in operation["parameters"] + if parameter["name"] == "X-Ux-Session-Id" + ) + assert ux_header["in"] == "header" + assert ux_header["required"] is True + + +@pytest.mark.asyncio +async def test_user_audit_context_requires_existing_owned_session() -> None: + session_id = uuid.uuid4() + user = SimpleNamespace(id=uuid.uuid4()) + incoming = request( + forwarded="203.0.113.10", + ux_session_id=str(session_id), + user_agent="Example Browser/1.0", + ) + incoming.state.request_id = str(uuid.uuid4()) + incoming.state.trace_id = "c" * 32 + incoming.state.user_agent_hash = user_agent_hash(incoming) + + class Db: + async def scalar(self, _query): + return session_id + + context = await required_user_audit_context(incoming, Db(), user, str(session_id)) + assert context.ux_session_id == session_id + assert context.client_ip == "203.0.113.10" + + missing = request(forwarded="203.0.113.10") + missing.state.request_id = str(uuid.uuid4()) + missing.state.trace_id = "d" * 32 + missing.state.user_agent_hash = None + with pytest.raises(DomainError, match="X-Ux-Session-Id is required"): + await required_user_audit_context(missing, Db(), user, "") diff --git a/codebase/backend/bitrix-local-app/app/main.py b/codebase/backend/bitrix-local-app/app/main.py index 6ed9234..b78c938 100644 --- a/codebase/backend/bitrix-local-app/app/main.py +++ b/codebase/backend/bitrix-local-app/app/main.py @@ -159,7 +159,7 @@ def retry_delay(attempt: int, maximum: int) -> float: def safely_retryable(exc: Exception) -> bool: - return isinstance(exc, httpx.ConnectError) or ( + return isinstance(exc, (httpx.ConnectError, httpx.TimeoutException)) or ( isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code in {429, 502, 503, 504} ) @@ -721,6 +721,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: request_fingerprint=fp, payload_json=body, status="sending", + lease_until=now() + timedelta(seconds=cfg.bitrix_http_timeout_sec + 5), ) session.add(row) try: @@ -775,6 +776,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: else: current.status = "ambiguous" current.last_error_code = "bitrix_delivery_ambiguous" + current.lease_until = None await session.commit() raise HTTPException( 503, @@ -1101,7 +1103,12 @@ async def reconcile_setup(app: FastAPI) -> dict[str, bool]: return result -async def claim_one(session: AsyncSession, model, statuses: list[str]): +async def claim_one( + session: AsyncSession, + model, + statuses: list[str], + claimed_status: str | None = None, +): row = await session.scalar( select(model) .where( @@ -1114,6 +1121,8 @@ async def claim_one(session: AsyncSession, model, statuses: list[str]): .limit(1) ) if row: + if claimed_status is not None: + row.status = claimed_status row.lease_until = now() + timedelta(seconds=30) row.attempt_count += 1 await session.commit() @@ -1239,7 +1248,12 @@ async def process_ack(app: FastAPI) -> None: async def process_outbound(app: FastAPI) -> None: async with app.state.sessions() as session: - row = await claim_one(session, OutboundMessage, ["retry"]) + row = await claim_one( + session, + OutboundMessage, + ["retry"], + claimed_status="sending", + ) if not row: return try: diff --git a/codebase/backend/bitrix-local-app/tests/test_core.py b/codebase/backend/bitrix-local-app/tests/test_core.py index 0295873..2d54338 100644 --- a/codebase/backend/bitrix-local-app/tests/test_core.py +++ b/codebase/backend/bitrix-local-app/tests/test_core.py @@ -26,6 +26,7 @@ from app.main import ( normalize_event, resolve_inbound_file_urls, retry_delay, + safely_retryable, validate_portal, ) @@ -206,6 +207,11 @@ def test_fingerprint_ignores_signed_query_and_portal_validation(): assert 0 <= retry_delay(4, 300) <= 8 +def test_network_timeouts_are_retryable(): + assert safely_retryable(httpx.ReadTimeout("Bitrix response timed out")) + assert safely_retryable(httpx.ConnectTimeout("Bitrix connection timed out")) + + @pytest.mark.asyncio async def test_bitrix_call_refreshes_and_retries_once_after_401(): auth_values: list[str] = [] diff --git a/codebase/backend/frontend-test-site/src/app-context.tsx b/codebase/backend/frontend-test-site/src/app-context.tsx index e0d2deb..e567708 100644 --- a/codebase/backend/frontend-test-site/src/app-context.tsx +++ b/codebase/backend/frontend-test-site/src/app-context.tsx @@ -43,9 +43,9 @@ export function AppProvider({ children }: { children: React.ReactNode }) { if (minutes > 0) setIdleTimeoutMs(minutes * 60_000); }).catch(() => undefined); void refreshTokens() - .then(() => { + .then(async () => { + await authApi.startSession("cold_start"); setAuthStatus("authenticated"); - void authApi.startSession("cold_start").catch(() => undefined); }) .catch(() => setAuthStatus("guest")); }, [toGuest]); diff --git a/codebase/backend/nginx/templates/site-tls.conf.template b/codebase/backend/nginx/templates/site-tls.conf.template index 4c298d3..ca2ce9a 100644 --- a/codebase/backend/nginx/templates/site-tls.conf.template +++ b/codebase/backend/nginx/templates/site-tls.conf.template @@ -107,8 +107,10 @@ server { include /etc/nginx/snippets/proxy-keycloak.conf; proxy_pass http://keycloak_upstream; } - location ~ ^/auth/realms/[^/]+/protocol/openid-connect/3p-cookies/ { + location ^~ /auth/realms/ { + limit_req zone=auth burst=10; include /etc/nginx/snippets/proxy-keycloak.conf; + proxy_read_timeout 60s; proxy_pass http://keycloak_upstream; } location /auth/ { diff --git a/codebase/backend/tests/test_config.py b/codebase/backend/tests/test_config.py index 95868d1..5cc0ef4 100644 --- a/codebase/backend/tests/test_config.py +++ b/codebase/backend/tests/test_config.py @@ -92,7 +92,7 @@ class InfrastructureConfigTests(unittest.TestCase): self.assertIn("include /etc/nginx/snippets/proxy-keycloak.conf;", site) self.assertIn("location = /auth/callback", site) self.assertIn("location ^~ /auth/resources/", site) - self.assertIn("protocol/openid-connect/3p-cookies/", site) + self.assertIn("location ^~ /auth/realms/", site) self.assertNotIn("security-headers.conf", proxy_keycloak) self.assertNotIn("X-Frame-Options", proxy_keycloak) @@ -185,7 +185,15 @@ class InfrastructureConfigTests(unittest.TestCase): self.assertIn("public.digest(", initial) self.assertIn("public.digest(", fix) self.assertIn('down_revision: str | None = "0001_initial"', fix) - self.assertIn('revision != "0002_pgcrypto_digest"', main) + self.assertIn('revision != "0003_consent_audit"', main) + + def test_consent_audit_migration_supports_existing_and_fresh_databases(self) -> None: + migration = ( + ROOT / "api-backend/alembic/versions/0003_consent_device_audit_context.py" + ).read_text(encoding="utf-8") + self.assertIn("ADD COLUMN IF NOT EXISTS device_json", migration) + self.assertIn("DROP COLUMN IF EXISTS ip", migration) + self.assertIn('down_revision: str | None = "0002_pgcrypto_digest"', migration) def test_keycloak_management_health_and_bridge_environment(self) -> None: standalone = (ROOT / "keycloak/docker-compose.yml").read_text(encoding="utf-8")