diff --git a/codebase/backend/api-backend/app/integrations.py b/codebase/backend/api-backend/app/integrations.py index cfba639..1428b6f 100644 --- a/codebase/backend/api-backend/app/integrations.py +++ b/codebase/backend/api-backend/app/integrations.py @@ -229,6 +229,7 @@ class S3Client: aws_access_key_id=settings.selectel_s3_access_key.get_secret_value(), aws_secret_access_key=settings.selectel_s3_secret_key.get_secret_value(), config=Config( + signature_version="s3v4", connect_timeout=3, read_timeout=10, retries={"max_attempts": 2}, diff --git a/codebase/backend/api-backend/app/realtime.py b/codebase/backend/api-backend/app/realtime.py index 8b8b5ba..d7e23ac 100644 --- a/codebase/backend/api-backend/app/realtime.py +++ b/codebase/backend/api-backend/app/realtime.py @@ -7,7 +7,7 @@ from typing import Any from redis.asyncio import Redis -CHANNEL_PREFIX = "han:realtime:dialog:" +CHANNEL_PREFIX = "han:rt:dialog:" class LocalFanout: diff --git a/codebase/backend/api-backend/tests/contract/test_clients.py b/codebase/backend/api-backend/tests/contract/test_clients.py index b4c4b6b..22b7996 100644 --- a/codebase/backend/api-backend/tests/contract/test_clients.py +++ b/codebase/backend/api-backend/tests/contract/test_clients.py @@ -1,5 +1,5 @@ import uuid -from urllib.parse import urlsplit +from urllib.parse import parse_qs, urlsplit import httpx import pytest @@ -11,6 +11,7 @@ from app.integrations import ( SafetyClient, fresh_openlines_payload, ) +from app.realtime import CHANNEL_PREFIX from app.settings import Settings @@ -41,6 +42,10 @@ def settings() -> Settings: return Settings.model_validate(common) +def test_realtime_channel_matches_redis_acl_namespace() -> None: + assert CHANNEL_PREFIX == "han:rt:dialog:" + + @pytest.mark.asyncio async def test_s3_presigned_urls_use_virtual_hosted_addressing() -> None: s3 = S3Client(settings()) @@ -52,6 +57,8 @@ async def test_s3_presigned_urls_use_virtual_hosted_addressing() -> None: assert urlsplit(put_url).path == "/quarantine/users/u/file.pdf" assert urlsplit(get_url).netloc == "attachments.s3.example" assert urlsplit(get_url).path == "/dialogs/d/file.pdf" + assert parse_qs(urlsplit(put_url).query)["X-Amz-Algorithm"] == ["AWS4-HMAC-SHA256"] + assert parse_qs(urlsplit(get_url).query)["X-Amz-Algorithm"] == ["AWS4-HMAC-SHA256"] @pytest.mark.asyncio diff --git a/codebase/backend/bitrix-local-app/app/main.py b/codebase/backend/bitrix-local-app/app/main.py index 9f69fbe..6ed9234 100644 --- a/codebase/backend/bitrix-local-app/app/main.py +++ b/codebase/backend/bitrix-local-app/app/main.py @@ -6,6 +6,7 @@ import hashlib import hmac import json import logging +import mimetypes import re import secrets import uuid @@ -259,18 +260,48 @@ def normalize_event(payload: dict[str, Any]) -> dict[str, Any] | None: files_raw = first(data, ("MESSAGES", "0", "message", "files"), ("MESSAGE", "FILES")) or [] if isinstance(files_raw, dict): files_raw = list(files_raw.values()) - files = [ - { - "name": str(item.get("name") or item.get("NAME") or "attachment"), - "mime_type": str( - item.get("type") or item.get("TYPE") or "application/octet-stream" - ), + files = [] + for item in files_raw: + if not isinstance(item, dict): + continue + name = str(item.get("name") or item.get("NAME") or "attachment") + declared_type = str( + item.get("mimeType") or item.get("mime_type") or item.get("mime") or "" + ) + if not declared_type: + candidate = str(item.get("type") or item.get("TYPE") or "") + declared_type = candidate if "/" in candidate else "" + guessed_type, _ = mimetypes.guess_type(name) + normalized_file = { + "name": name, + "mime_type": declared_type or guessed_type or "application/octet-stream", "size_bytes": int(item.get("size") or item.get("SIZE") or 0), - "download_url": str(item.get("url") or item.get("URL") or ""), + "download_url": str( + item.get("url") + or item.get("URL") + or item.get("urlDownload") + or item.get("downloadUrl") + or item.get("downloadLink") + or item.get("link") + or item.get("URL_DOWNLOAD") + or "" + ), } - for item in files_raw - if isinstance(item, dict) - ] + file_id = ( + item.get("id") + or item.get("ID") + or item.get("fileId") + or item.get("file_id") + or item.get("FILE_ID") + ) + if file_id: + normalized_file["_bitrix_file_id"] = str(file_id) + if not normalized_file["download_url"] and not file_id: + logger.warning( + "inbound file has no supported download reference; keys=%s", + sorted(str(key) for key in item), + ) + files.append(normalized_file) if not text_value and not files: raise ValueError("empty message") message = {"text": text_value, "files": files} @@ -302,8 +333,14 @@ class BitrixClient: self.limit = asyncio.Semaphore(settings.bitrix_http_max_concurrency) self.refresh_lock = asyncio.Lock() - async def ensure_fresh(self, portal: PortalInstallation) -> None: - if portal.expires_at and portal.expires_at > now() + timedelta(seconds=60): + async def ensure_fresh( + self, + portal: PortalInstallation, + *, + force: bool = False, + stale_access_ciphertext: str | None = None, + ) -> None: + if not force and portal.expires_at and portal.expires_at > now() + timedelta(seconds=60): return async with self.refresh_lock: async with self.sessions() as session: @@ -314,7 +351,17 @@ class BitrixClient: await session.commit() try: current = await session.get(PortalInstallation, portal.id) - if current.expires_at and current.expires_at > now() + timedelta(seconds=60): + refreshed_by_peer = ( + force + and stale_access_ciphertext is not None + and current.access_ciphertext != stale_access_ciphertext + ) + still_fresh = ( + not force + and current.expires_at + and current.expires_at > now() + timedelta(seconds=60) + ) + if refreshed_by_peer or still_fresh: portal.access_ciphertext = current.access_ciphertext portal.access_nonce = current.access_nonce portal.expires_at = current.expires_at @@ -370,18 +417,30 @@ class BitrixClient: async def call(self, portal: PortalInstallation, method: str, fields: dict[str, Any]) -> dict: await self.ensure_fresh(portal) - token = self.cipher.decrypt( - portal.access_ciphertext, - portal.access_nonce, - portal.member_id, - portal.domain, - "access", - ) url = f"https://{portal.domain}/rest/{method}.json" - async with self.limit: - response = await self.http.post( - url, data={**fields, "auth": token}, follow_redirects=False + + async def post() -> httpx.Response: + token = self.cipher.decrypt( + portal.access_ciphertext, + portal.access_nonce, + portal.member_id, + portal.domain, + "access", ) + async with self.limit: + return await self.http.post( + url, data={**fields, "auth": token}, follow_redirects=False + ) + + stale_access_ciphertext = portal.access_ciphertext + response = await post() + if response.status_code == 401: + await self.ensure_fresh( + portal, + force=True, + stale_access_ciphertext=stale_access_ciphertext, + ) + response = await post() response.raise_for_status() body = response.json() if body.get("error"): @@ -693,6 +752,14 @@ def create_app(settings: Settings | None = None) -> FastAPI: try: result = await deliver_outbound(request.app, row.id) except Exception as exc: + logger.exception( + "outbound delivery failed", + extra={ + "request_id": request.state.request_id, + "message_id": str(dto.message_id), + "error_type": type(exc).__name__, + }, + ) async with request.app.state.sessions() as session: current = await session.get(OutboundMessage, row.id, with_for_update=True) if current: @@ -1078,9 +1145,11 @@ async def process_inbox(app: FastAPI) -> None: if not row: return try: + payload = json.loads(json.dumps(row.normalized_json)) + await resolve_inbound_file_urls(app, payload) response = await app.state.http.post( app.state.settings.bitrix_api_forward_url, - json=row.normalized_json, + json=payload, headers={ "Authorization": f"Bearer {app.state.settings.bitrix_api_forward_token}", "X-Request-ID": str(uuid.uuid4()), @@ -1105,10 +1174,40 @@ async def process_inbox(app: FastAPI) -> None: ) ) await session.commit() - except Exception: + except Exception as exc: + logger.exception( + "inbound forward failed", + extra={ + "inbox_id": str(row.id), + "error_type": type(exc).__name__, + }, + ) await mark_retry(app, InboxEvent, row.id, "api_forward_failed") +async def resolve_inbound_file_urls(app: FastAPI, payload: dict[str, Any]) -> None: + message = payload.get("message") or {} + for item in message.get("files") or []: + file_id = item.pop("_bitrix_file_id", "") + if item.get("download_url"): + continue + if not file_id: + raise RuntimeError("bitrix_file_id_missing") + async with app.state.sessions() as session: + portal = await active_portal(session) + if not portal: + raise RuntimeError("portal_not_installed") + result = await app.state.bitrix.call( + portal, + "im.v2.File.download", + {"id": file_id}, + ) + download_url = result.get("downloadUrl") or result.get("urlDownload") + if not download_url: + raise RuntimeError("bitrix_file_download_url_missing") + item["download_url"] = str(download_url) + + async def process_ack(app: FastAPI) -> None: async with app.state.sessions() as session: row = await claim_one(session, DeliveryAckOutbox, ["pending", "retry"]) @@ -1145,7 +1244,14 @@ async def process_outbound(app: FastAPI) -> None: return try: await deliver_outbound(app, row.id) - except Exception: + except Exception as exc: + logger.exception( + "outbound retry failed", + extra={ + "outbound_id": str(row.id), + "error_type": type(exc).__name__, + }, + ) await mark_retry(app, OutboundMessage, row.id, "bitrix_delivery_failed") diff --git a/codebase/backend/bitrix-local-app/tests/test_core.py b/codebase/backend/bitrix-local-app/tests/test_core.py index 9272007..0295873 100644 --- a/codebase/backend/bitrix-local-app/tests/test_core.py +++ b/codebase/backend/bitrix-local-app/tests/test_core.py @@ -1,6 +1,8 @@ import base64 import os import uuid +from types import SimpleNamespace +from urllib.parse import parse_qs os.environ.setdefault("BITRIX_DATABASE_URL", "postgresql://unused/unused") os.environ.setdefault("BITRIX_CLIENT_ID", "client") @@ -14,12 +16,15 @@ os.environ.setdefault( base64.urlsafe_b64encode(b"k" * 32).decode().rstrip("="), ) +import httpx import pytest from app.main import ( + BitrixClient, TokenCipher, canonical_fingerprint, normalize_event, + resolve_inbound_file_urls, retry_delay, validate_portal, ) @@ -58,6 +63,135 @@ def test_normalize_message_and_finish(): assert closed["event_type"] == "dialog.closed" +def test_normalize_bitrix_file_uses_download_url_and_infers_mime_type(): + external = str(uuid.uuid4()) + message = normalize_event( + { + "event": "ONIMCONNECTORMESSAGEADD", + "data": { + "MESSAGES": [ + { + "im": {"chat_id": 1807, "message_id": 86498}, + "chat": {"id": external}, + "message": { + "text": "", + "files": [ + { + "name": "image.png", + "type": "image", + "size": 941380, + "urlDownload": "https://portal.bitrix24.ru/download/file.png", + } + ], + }, + } + ] + }, + } + ) + + assert message["message"]["files"] == [ + { + "name": "image.png", + "mime_type": "image/png", + "size_bytes": 941380, + "download_url": "https://portal.bitrix24.ru/download/file.png", + } + ] + + +def test_normalize_bitrix_file_uses_open_lines_download_link_and_mime(): + external = str(uuid.uuid4()) + message = normalize_event( + { + "event": "ONIMCONNECTORMESSAGEADD", + "data": { + "MESSAGES": [ + { + "im": {"chat_id": 1807, "message_id": 86499}, + "chat": {"id": external}, + "message": { + "text": "", + "files": [ + { + "name": "diploma.jpg", + "type": "image", + "mime": "image/jpeg", + "size": 236934, + "downloadLink": ( + "https://portal.bitrix24.ru/download/diploma.jpg" + ), + "link": "https://portal.bitrix24.ru/view/diploma.jpg", + "width": 1200, + "height": 800, + } + ], + }, + } + ] + }, + } + ) + + assert message["message"]["files"] == [ + { + "name": "diploma.jpg", + "mime_type": "image/jpeg", + "size_bytes": 236934, + "download_url": "https://portal.bitrix24.ru/download/diploma.jpg", + } + ] + + +@pytest.mark.asyncio +async def test_resolve_inbound_file_url_from_bitrix_file_id(monkeypatch): + class SessionContext: + async def __aenter__(self): + return object() + + async def __aexit__(self, *_args): + return None + + class Bitrix: + async def call(self, portal, method, params): + assert portal == "portal" + assert method == "im.v2.File.download" + assert params == {"id": "5155"} + return {"downloadUrl": "https://portal.bitrix24.ru/download/file.png"} + + async def fake_active_portal(_session): + return "portal" + + monkeypatch.setattr("app.main.active_portal", fake_active_portal) + app = SimpleNamespace( + state=SimpleNamespace(sessions=lambda: SessionContext(), bitrix=Bitrix()) + ) + payload = { + "message": { + "files": [ + { + "name": "image.png", + "mime_type": "image/png", + "size_bytes": 941380, + "download_url": "", + "_bitrix_file_id": "5155", + } + ] + } + } + + await resolve_inbound_file_urls(app, payload) + + assert payload["message"]["files"] == [ + { + "name": "image.png", + "mime_type": "image/png", + "size_bytes": 941380, + "download_url": "https://portal.bitrix24.ru/download/file.png", + } + ] + + def test_fingerprint_ignores_signed_query_and_portal_validation(): payload = {"message": {"files": [{"download_url": "https://s3/object?sig=one"}]}} other = {"message": {"files": [{"download_url": "https://s3/object?sig=two"}]}} @@ -70,3 +204,46 @@ def test_fingerprint_ignores_signed_query_and_portal_validation(): with pytest.raises(ValueError): validate_portal("evil.example", "https://evil.example/rest/", "han0107.bitrix24.ru") assert 0 <= retry_delay(4, 300) <= 8 + + +@pytest.mark.asyncio +async def test_bitrix_call_refreshes_and_retries_once_after_401(): + auth_values: list[str] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + auth = parse_qs(request.content.decode())["auth"][0] + auth_values.append(auth) + if auth == "old-access": + return httpx.Response(401, request=request) + return httpx.Response(200, json={"result": {"ok": True}}, request=request) + + class Cipher: + @staticmethod + def decrypt(ciphertext, *_): + return ciphertext + + portal = SimpleNamespace( + access_ciphertext="old-access", + access_nonce="nonce", + member_id="member", + domain="han0107.bitrix24.ru", + expires_at=None, + ) + settings = SimpleNamespace(bitrix_http_max_concurrency=2) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: + client = BitrixClient(http, settings, Cipher(), sessions=None) + refresh_calls: list[bool] = [] + + async def ensure_fresh(_, *, force=False, stale_access_ciphertext=None): + refresh_calls.append(force) + if force: + assert stale_access_ciphertext == "old-access" + portal.access_ciphertext = "new-access" + + client.ensure_fresh = ensure_fresh + result = await client.call(portal, "imconnector.send.messages", {"MESSAGE": "test"}) + + assert result == {"ok": True} + assert auth_values == ["old-access", "new-access"] + assert refresh_calls == [False, True]