diff --git a/CHANGELOG.md b/CHANGELOG.md index d94f852c3..4f1010d8e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,7 @@ ## [Unreleased] +- 이메일 발송 연결 중 요청을 취소하면 남은 연결을 즉시 정리합니다. 직접 발송과 등록된 Connector에 같은 수정이 적용되며, 취소를 발송 성공으로 처리하지 않습니다. +- 이메일 발송 제한을 모든 worker가 공유하는 PostgreSQL rolling window로 + 적용하고, 제한 상태를 확인할 수 없으면 발송 전에 안전하게 중단합니다. - 데이터 저장·검색 검사가 실제 DB 연결 없이 건너뛰어져도 성공으로 보이던 검증 공백을 보완했습니다. 새 DB 설치와 반복 업그레이드 후 전체 백엔드 검사를 실행하며, 아직 실제 배포 환경 검증을 뜻하지는 않습니다. - Starlette `TestClient`의 기존 `httpx2==2.5.0` pin을 core 개발·테스트 의존성으로 승격하고, deprecated `httpx` fallback 경고 억제를 제거했습니다. - 긴 이메일·첨부 본문을 의미 단위 청크로 임베딩한 뒤 기존 email/attachment 벡터 계약으로 평균화하고, 청크 요청·벡터 누적을 제한된 창으로 처리합니다. OpenAI `text-embedding-3-*`에는 저장 차원(`1536`)을 직접 요청하도록 보강했습니다. 합성 메일 fixture 5건(70청크)과 provider 요청 계약으로 1,536차원 벡터 경로를 검증했으며, 실행 시 선택한 임베딩 제공자에 본문·파싱된 첨부 텍스트를 전송할 수 있습니다. 회사 기밀 데이터는 fixture·commit·PR·log에 포함하지 않습니다. diff --git a/backend/alembic/versions/0018_security_audit_events.py b/backend/alembic/versions/0018_security_audit_events.py new file mode 100644 index 000000000..0fccdca5c --- /dev/null +++ b/backend/alembic/versions/0018_security_audit_events.py @@ -0,0 +1,84 @@ +"""Formalize durable security audit events in the Alembic upgrade path. + +Revision ID: 0018_security_audit_events +Revises: 0017_merge_newsdom_carddav_heads +Create Date: 2026-09-05 00:00:00.000000 + +Older installations can already contain ``security_audit_events`` because the +legacy bootstrap path created it outside Alembic. This revision is therefore +idempotent: it creates the table when absent and reconciles the model-owned +indexes when the table already exists. Downgrade intentionally preserves the +durable audit table and its evidence rather than deleting security history that +may predate this revision. +""" + +from alembic import op +import sqlalchemy as sa + +revision = "0018_security_audit_events" +down_revision = "0017_merge_newsdom_carddav_heads" +branch_labels = None +depends_on = None + +_TABLE = "security_audit_events" +_INDEXES: tuple[tuple[str, list[str]], ...] = ( + ("ix_security_audit_events_actor_user_id", ["actor_user_id"]), + ("ix_security_audit_events_actor_role", ["actor_role"]), + ("ix_security_audit_events_organization_id", ["organization_id"]), + ("ix_security_audit_events_workspace_id", ["workspace_id"]), + ("ix_security_audit_events_event_action", ["event_action"]), + ("ix_security_audit_events_resource_type", ["resource_type"]), + ("ix_security_audit_events_resource_uid", ["resource_uid"]), + ("ix_security_audit_events_observed_at", ["observed_at"]), + ( + "ix_security_audit_events_scope_time", + ["organization_id", "workspace_id", "observed_at"], + ), + ( + "ix_security_audit_events_actor_scope", + ["actor_user_id", "organization_id", "workspace_id"], + ), +) + + +def upgrade() -> None: + """Create the audit schema missing from Alembic-managed upgrades.""" + connection = op.get_bind() + inspector = sa.inspect(connection) + + if not inspector.has_table(_TABLE): + op.create_table( + _TABLE, + sa.Column("event_uid", sa.String(), nullable=False), + sa.Column("actor_user_id", sa.String(), nullable=False), + sa.Column("actor_role", sa.String(), nullable=False), + sa.Column("organization_id", sa.String(), nullable=True), + sa.Column("workspace_id", sa.String(), nullable=False), + sa.Column("event_action", sa.String(), nullable=False), + sa.Column("resource_type", sa.String(), nullable=False), + sa.Column("resource_uid", sa.String(), nullable=True), + sa.Column("evidence_source", sa.String(), nullable=False), + sa.Column("detail_text", sa.Text(), nullable=True), + sa.Column( + "observed_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + sa.PrimaryKeyConstraint("event_uid"), + ) + + for index_name, column_names in _INDEXES: + op.create_index( + index_name, + _TABLE, + column_names, + if_not_exists=True, + ) + + +def downgrade() -> None: + """Preserve durable security evidence created before or after this revision.""" + # This revision reconciles a table that may predate Alembic ownership. + # Dropping it on downgrade could destroy security evidence owned by the + # earlier bootstrap path, so schema rollback deliberately leaves it intact. diff --git a/backend/alembic/versions/0020_merge_send_registry.py b/backend/alembic/versions/0020_merge_send_registry.py new file mode 100644 index 000000000..71c27300b --- /dev/null +++ b/backend/alembic/versions/0020_merge_send_registry.py @@ -0,0 +1,18 @@ +"""Join shared-send audit and workspace registry migration histories. + +Both parents own durable data. This revision only reconciles the graph so the +normal ``upgrade head`` path applies both prerequisites without deleting either. +""" + +revision = "0020_merge_send_registry" +down_revision = ("0018_security_audit_events", "0019_email_read_state_repair") +branch_labels = None +depends_on = None + + +def upgrade() -> None: + """Join already-applied parent revisions without changing their data.""" + + +def downgrade() -> None: + """Split revision bookkeeping without deleting parent-owned data.""" diff --git a/backend/api/emails.py b/backend/api/emails.py index 2b0a9dbd6..fcef299ad 100644 --- a/backend/api/emails.py +++ b/backend/api/emails.py @@ -1,5 +1,4 @@ from collections import defaultdict -from threading import Lock from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import func, or_, select @@ -7,7 +6,6 @@ from db.models import Email from pydantic import BaseModel, EmailStr, Field, field_validator import datetime -import time from typing import Literal from services.email_client import ( EmailMessageParams, @@ -39,6 +37,10 @@ canonical_email_import_upload_filename, import_email_uploads, ) +from services.email_send_rate_limiter import ( + EmailSendRateLimitUnavailable, + enforce_send_email_rate_limit, +) from services.llm_provider_selection import resolve_runtime_llm_provider from services.text_safety import strip_html_markup import logging @@ -50,34 +52,6 @@ router = APIRouter(prefix="/api/emails") -_SEND_EMAIL_RATE_LIMIT_MAX_ATTEMPTS = 10 -_SEND_EMAIL_RATE_LIMIT_WINDOW_SECONDS = 60.0 -_email_send_attempts_by_scope: dict[tuple[str | None, str], list[float]] = {} -_email_send_rate_limit_lock = Lock() - - -def _enforce_send_email_rate_limit(auth_context: AuthContext) -> None: - now = time.monotonic() - cutoff = now - _SEND_EMAIL_RATE_LIMIT_WINDOW_SECONDS - key = (auth_context.organization_id, auth_context.user_id) - - # ponytail: process-local throttle; move to Redis when multi-worker send volume matters. - with _email_send_rate_limit_lock: - attempts = [ - attempt - for attempt in _email_send_attempts_by_scope.get(key, []) - if attempt > cutoff - ] - if len(attempts) >= _SEND_EMAIL_RATE_LIMIT_MAX_ATTEMPTS: - _email_send_attempts_by_scope[key] = attempts - raise HTTPException( - status_code=429, - detail="Email send rate limit exceeded", - ) - attempts.append(now) - _email_send_attempts_by_scope[key] = attempts - - def canonical_thread_key(email: Email) -> str: return ( normalize_message_id(email.thread_id) @@ -729,6 +703,11 @@ async def send_email_endpoint( smtp_port = tenant_config.smtp_port smtp_username = tenant_config.smtp_username smtp_password = tenant_config.smtp_password + # The production dependency is an AsyncSession. Lightweight unit-test + # doubles do not own a pooled connection and may omit rollback(). + rollback = getattr(db, "rollback", None) + if rollback is not None: + await rollback() validate_smtp_destination(smtp_server, smtp_port) except Exception as exc: if "ENCRYPTION_KEY is required" in str(exc): @@ -756,7 +735,18 @@ async def send_email_endpoint( in_reply_to=request.in_reply_to, references=request.references, ) - _enforce_send_email_rate_limit(auth_context) + try: + rate_limit_decision = await enforce_send_email_rate_limit(auth_context) + except EmailSendRateLimitUnavailable as exc: + raise HTTPException( + status_code=503, + detail="Email send rate limiter unavailable", + ) from exc + if not rate_limit_decision.allowed: + raise HTTPException( + status_code=429, + detail="Email send rate limit exceeded", + ) smtp_config = SmtpConfig( smtp_server=smtp_server, smtp_port=smtp_port, diff --git a/backend/services/email_client.py b/backend/services/email_client.py index 8763a41aa..e82e81c33 100644 --- a/backend/services/email_client.py +++ b/backend/services/email_client.py @@ -400,7 +400,7 @@ async def _connect_validated_smtp_socket( ), timeout=SMTP_TIMEOUT_SECONDS, ) - except Exception: + except (Exception, asyncio.CancelledError): smtp_socket.close() raise return smtp_socket diff --git a/backend/services/email_send_rate_limiter.py b/backend/services/email_send_rate_limiter.py new file mode 100644 index 000000000..50d2dbeac --- /dev/null +++ b/backend/services/email_send_rate_limiter.py @@ -0,0 +1,186 @@ +"""Shared, fail-closed email send throttling.""" + +from __future__ import annotations + +import datetime +import hashlib +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING, Literal + +from sqlalchemy import bindparam, delete, func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from db.models import SecurityAuditEvent +from db.session import AsyncSessionLocal + +if TYPE_CHECKING: + from api.auth import AuthContext + +logger = logging.getLogger(__name__) + +SEND_RATE_LIMIT_MAX_ATTEMPTS = 10 +SEND_RATE_LIMIT_WINDOW_SECONDS = 60 +SEND_RATE_LIMIT_NAMESPACE = "naruon-email-send-rate-limit" + + +class EmailSendRateLimitUnavailable(RuntimeError): + """The shared rate-limit state cannot provide a trustworthy decision.""" + + +@dataclass(frozen=True) +class EmailSendRateLimitDecision: + """A non-sensitive rate-limit decision returned to the send endpoint.""" + + allowed: bool + reason: Literal["allowed", "quota_exhausted"] + + +def rate_limit_scope_hash( + user_id: str, organization_id: str | None, workspace_id: str +) -> str: + """Return a non-reversible identifier for one authorized send scope.""" + organization_scope = organization_id or "" + value = ( + f"{SEND_RATE_LIMIT_NAMESPACE}\0{organization_scope}\0{workspace_id}\0{user_id}" + ) + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _lock_key(scope_hash: str) -> int: + return int.from_bytes(bytes.fromhex(scope_hash[:16]), byteorder="big", signed=True) + + +def _session_uses_postgresql(session: AsyncSession) -> bool: + try: + bind = session.get_bind() + except Exception: + return False + return getattr(getattr(bind, "dialect", None), "name", None) == "postgresql" + + +def _audit_event( + auth_context: AuthContext, + *, + scope_hash: str, + decision: EmailSendRateLimitDecision, + observed_at: datetime.datetime, +) -> SecurityAuditEvent: + return SecurityAuditEvent( + actor_user_id=auth_context.user_id, + actor_role=auth_context.role, + organization_id=auth_context.organization_id, + workspace_id=auth_context.workspace_id, + event_action=f"email_send_rate_limit.{decision.reason}", + resource_type="email_send_rate_limit", + resource_uid=f"email_send_scope:{scope_hash}", + evidence_source="services.email_send_rate_limiter", + observed_at=observed_at, + detail_text=( + f"decision={decision.reason};" + f"window_seconds={SEND_RATE_LIMIT_WINDOW_SECONDS};" + f"max_attempts={SEND_RATE_LIMIT_MAX_ATTEMPTS}" + ), + ) + + +async def enforce_send_email_rate_limit( + auth_context: AuthContext, + *, + now: datetime.datetime | None = None, +) -> EmailSendRateLimitDecision: + """Atomically reserve one send attempt in a rolling PostgreSQL window. + + A limiter-owned transaction prevents committing unrelated request work. + PostgreSQL advisory locking serializes the count-and-record decision across + workers. Production timestamps come from PostgreSQL after the scope lock so + worker clock skew and lock wait do not weaken the real-time quota. Expired + allowed rows are transient reservation state and are pruned under that same + lock; durable quota-denial audit evidence is retained. ``now`` remains an + explicit deterministic test seam. + """ + observed_at = now + scope_hash = rate_limit_scope_hash( + auth_context.user_id, + auth_context.organization_id, + auth_context.workspace_id, + ) + scope_uid = f"email_send_scope:{scope_hash}" + async with AsyncSessionLocal() as session: + if not _session_uses_postgresql(session): + raise EmailSendRateLimitUnavailable + try: + await session.connection( + execution_options={"isolation_level": "READ COMMITTED"} + ) + await session.execute( + select(func.pg_advisory_xact_lock(bindparam("lock_key"))), + {"lock_key": _lock_key(scope_hash)}, + ) + if observed_at is None: + database_clock = await session.execute(select(func.clock_timestamp())) + observed_at = database_clock.scalar_one() + window_started_at = observed_at - datetime.timedelta( + seconds=SEND_RATE_LIMIT_WINDOW_SECONDS + ) + await session.execute( + delete(SecurityAuditEvent).where( + SecurityAuditEvent.resource_uid == scope_uid, + SecurityAuditEvent.event_action + == "email_send_rate_limit.allowed", + SecurityAuditEvent.observed_at <= window_started_at, + ) + ) + result = await session.execute( + select(func.count()) + .select_from(SecurityAuditEvent) + .where( + SecurityAuditEvent.resource_uid == scope_uid, + SecurityAuditEvent.event_action + == "email_send_rate_limit.allowed", + SecurityAuditEvent.observed_at > window_started_at, + ) + ) + allowed = result.scalar_one() < SEND_RATE_LIMIT_MAX_ATTEMPTS + decision = EmailSendRateLimitDecision( + allowed=allowed, + reason="allowed" if allowed else "quota_exhausted", + ) + record_decision = allowed + if not allowed: + denied_result = await session.execute( + select(func.count()) + .select_from(SecurityAuditEvent) + .where( + SecurityAuditEvent.resource_uid == scope_uid, + SecurityAuditEvent.event_action + == "email_send_rate_limit.quota_exhausted", + SecurityAuditEvent.observed_at > window_started_at, + ) + ) + record_decision = denied_result.scalar_one() == 0 + if record_decision: + session.add( + _audit_event( + auth_context, + scope_hash=scope_hash, + decision=decision, + observed_at=observed_at, + ) + ) + await session.commit() + return decision + except Exception as exc: + try: + await session.rollback() + except Exception as rollback_exc: + logger.warning( + "Email send rate limiter rollback failed; error_type=%s", + type(rollback_exc).__name__, + ) + logger.warning( + "Email send rate limiter decision unavailable; " + "event_action=email_send_rate_limit.unavailable error_type=%s", + type(exc).__name__, + ) + raise EmailSendRateLimitUnavailable from exc diff --git a/backend/tests/test_email_client_smtp.py b/backend/tests/test_email_client_smtp.py index 7b2a185a5..5e1e4c0e8 100644 --- a/backend/tests/test_email_client_smtp.py +++ b/backend/tests/test_email_client_smtp.py @@ -4,6 +4,7 @@ import pytest import services.email_client as email_client +from runner.local_mail_adapters import LocalMailAccountConfig, LocalMailAdapters import os @@ -14,6 +15,72 @@ def _make_socket() -> socket.socket: return socket.socket(socket.AF_INET, socket.SOCK_STREAM) +@pytest.mark.asyncio +@pytest.mark.parametrize("send_path", ["direct_send", "registered_connector"]) +async def test_smtp_connect_cancellation_closes_socket(monkeypatch, send_path): + """Cancelling either sender must close its real socket before returning.""" + monkeypatch.setattr(email_client.settings, "ALLOWED_SMTP_HOSTS", "smtp.example.com") + monkeypatch.setattr( + email_client.socket, + "getaddrinfo", + lambda *args, **kwargs: [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 587)) + ], + ) + connect_started = asyncio.Event() + connect_pending = asyncio.Event() + captured_sockets = [] + + async def pending_connect(smtp_socket, socket_address): + assert socket_address == ("8.8.8.8", 587) + captured_sockets.append(smtp_socket) + connect_started.set() + await connect_pending.wait() + + monkeypatch.setattr(asyncio.get_running_loop(), "sock_connect", pending_connect) + smtp_config = email_client.SmtpConfig( + smtp_server="smtp.example.com", + smtp_port=587, + smtp_username="sender@example.com", + smtp_password="unit-test-value", + ) + if send_path == "direct_send": + send_operation = email_client.send_email( + email_client.EmailMessageParams("recipient@example.com", "Subject", "Body"), + smtp_config, + ) + else: + mail_adapters = LocalMailAdapters([ + LocalMailAccountConfig( + account="test_account", + user_id="test_user", + smtp_server=smtp_config.smtp_server, + smtp_port=smtp_config.smtp_port, + smtp_username=smtp_config.smtp_username, + smtp_password=smtp_config.smtp_password, + ) + ]) + send_operation = mail_adapters.send_smtp({ + "account": "test_account", + "to": "recipient@example.com", + "subject": "Subject", + "body": "Body", + }) + send_task = asyncio.create_task(send_operation) + try: + await asyncio.wait_for(connect_started.wait(), timeout=5) + send_task.cancel("send cancelled") + with pytest.raises(asyncio.CancelledError, match="send cancelled"): + await send_task + assert len(captured_sockets) == 1 + assert captured_sockets[0].fileno() == -1 + finally: + send_task.cancel() + await asyncio.gather(send_task, return_exceptions=True) + for smtp_socket in captured_sockets: + smtp_socket.close() + + def test_smtp_host_policy_denies_empty_allowlist_before_dns(monkeypatch): monkeypatch.setattr(email_client.settings, "ALLOWED_SMTP_HOSTS", "") diff --git a/backend/tests/test_email_send_connection_pool.py b/backend/tests/test_email_send_connection_pool.py new file mode 100644 index 000000000..d97961171 --- /dev/null +++ b/backend/tests/test_email_send_connection_pool.py @@ -0,0 +1,98 @@ +"""Regression coverage for bounded-pool email-send connection ordering.""" + +import asyncio +from types import SimpleNamespace + +import pytest + +from api import emails as emails_api +from api.auth import AuthContext +from services.email_send_rate_limiter import EmailSendRateLimitDecision + + +class _TenantResult: + def __init__(self, tenant_config): + self._tenant_config = tenant_config + + def scalar_one_or_none(self): + return self._tenant_config + + +class _SingleSlotRequestSession: + """Model a request session that holds the only pool slot after a read.""" + + def __init__(self): + self.connection_slot = asyncio.Semaphore(1) + self.holds_connection = False + self.events: list[str] = [] + self.tenant_config = SimpleNamespace( + smtp_server="smtp.example.com", + smtp_port=587, + smtp_username="testuser", + smtp_password=None, + ) + + async def execute(self, _query): + await self.connection_slot.acquire() + self.holds_connection = True + self.events.append("tenant-config-read") + return _TenantResult(self.tenant_config) + + async def rollback(self): + self.events.append("request-read-transaction-ended") + if self.holds_connection: + self.holds_connection = False + self.connection_slot.release() + + +@pytest.mark.asyncio +async def test_send_releases_request_read_connection_before_limiter_session(monkeypatch): + """A one-slot pool must not self-deadlock when the limiter opens its session.""" + + session = _SingleSlotRequestSession() + auth_context = AuthContext( + user_id="testuser", + role="member", + organization_id="org-acme", + group_ids=(), + workspace_id="workspace-org-acme", + ) + + monkeypatch.setattr( + emails_api, + "validate_smtp_destination", + lambda smtp_server, smtp_port, *, resolve_host=True: (smtp_server, smtp_port), + ) + + async def bounded_pool_rate_limit(_auth_context): + await asyncio.wait_for(session.connection_slot.acquire(), timeout=0.05) + session.events.append("limiter-session-acquired") + session.connection_slot.release() + return EmailSendRateLimitDecision(allowed=True, reason="allowed") + + async def fake_send_email(*, message_params, smtp_config): + assert message_params.to_address == "test@example.com" + assert smtp_config.smtp_server == "smtp.example.com" + return {"status": "sent", "simulated": False} + + monkeypatch.setattr( + emails_api, "enforce_send_email_rate_limit", bounded_pool_rate_limit + ) + monkeypatch.setattr(emails_api, "send_email", fake_send_email) + + result = await emails_api.send_email_endpoint( + emails_api.SendEmailRequest( + to="test@example.com", + subject="Pool regression", + body="Body", + ), + db=session, + auth_context=auth_context, + ) + + assert result == {"status": "sent", "simulated": False} + assert session.events == [ + "tenant-config-read", + "request-read-transaction-ended", + "limiter-session-acquired", + ] diff --git a/backend/tests/test_email_send_rate_limiter.py b/backend/tests/test_email_send_rate_limiter.py new file mode 100644 index 000000000..612b499f1 --- /dev/null +++ b/backend/tests/test_email_send_rate_limiter.py @@ -0,0 +1,642 @@ +import asyncio +import datetime +import time +import uuid +from types import SimpleNamespace + +import pytest +import pytest_asyncio +import jwt +from httpx import ASGITransport, AsyncClient + +from sqlalchemy import delete, func, select, text +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from api.auth import AuthContext +from api import emails as emails_api +from core.config import settings +from db.models import SecurityAuditEvent, TenantConfig +from db.session import get_db +from main import app +import services.email_send_rate_limiter as limiter_module + + +class _Result: + def __init__(self, row=0): + self.row = row + + def scalar_one(self): + return self.row + + +class _SharedAttemptStore: + def __init__(self): + self.events = [] + self.lock = asyncio.Lock() + + +class _PostgresSession: + def __init__(self, store, *, database_now: datetime.datetime | None = None): + self.store = store + self.added = [] + self.queries = [] + self.lock_held = False + self.pending = [] + self.isolation_level = None + self.database_now = database_now or datetime.datetime( + 2026, 8, 19, tzinfo=datetime.timezone.utc + ) + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + if self.lock_held: + self.store.lock.release() + self.lock_held = False + + def get_bind(self): + return SimpleNamespace(dialect=SimpleNamespace(name="postgresql")) + + async def connection(self, *, execution_options): + self.isolation_level = execution_options["isolation_level"] + return self + + async def execute(self, query, params=None): + query_text = str(query).lower() + self.queries.append(query_text) + if "pg_advisory_xact_lock" in query_text: + await self.store.lock.acquire() + self.lock_held = True + return _Result() + if "clock_timestamp" in query_text: + return _Result(self.database_now) + if query_text.startswith("delete from security_audit_events"): + values = tuple(query.compile().params.values()) + resource_uid = next( + value for value in values if str(value).startswith("email_send_scope:") + ) + event_action = next( + value + for value in values + if str(value).startswith("email_send_rate_limit.") + ) + cutoff = next( + value for value in values if isinstance(value, datetime.datetime) + ) + retained_events = [ + event + for event in self.store.events + if not ( + event.resource_uid == resource_uid + and event.event_action == event_action + and event.observed_at <= cutoff + ) + ] + deleted_count = len(self.store.events) - len(retained_events) + self.store.events = retained_events + return _Result(deleted_count) + if "security_audit_events" in query_text: + values = tuple(query.compile().params.values()) + resource_uid = next( + value for value in values if str(value).startswith("email_send_scope:") + ) + event_action = next( + value + for value in values + if str(value).startswith("email_send_rate_limit.") + ) + window_started_at = next( + value for value in values if isinstance(value, datetime.datetime) + ) + return _Result( + sum( + event.resource_uid == resource_uid + and event.event_action == event_action + and event.observed_at > window_started_at + for event in self.store.events + ) + ) + raise AssertionError(f"unexpected query: {query_text}") + + def add(self, item): + self.added.append(item) + if isinstance(item, SecurityAuditEvent): + self.pending.append(item) + + async def commit(self): + self.store.events.extend(self.pending) + self.pending.clear() + if self.lock_held: + self.store.lock.release() + self.lock_held = False + + async def rollback(self): + self.pending.clear() + if self.lock_held: + self.store.lock.release() + self.lock_held = False + + +def _context( + user_id="user-1", organization_id="org-1", workspace_id: str | None = None +): + return AuthContext( + user_id=user_id, + role="member", + organization_id=organization_id, + group_ids=(), + workspace_id=workspace_id or f"workspace-{organization_id or user_id}", + ) + + +@pytest.mark.asyncio +async def test_sliding_window_limits_concurrent_workers_without_cross_scope_leakage( + monkeypatch, +): + store = _SharedAttemptStore() + monkeypatch.setattr( + limiter_module, "AsyncSessionLocal", lambda: _PostgresSession(store) + ) + observed_at = datetime.datetime(2026, 8, 19, tzinfo=datetime.timezone.utc) + + async def attempt(user_id): + return await limiter_module.enforce_send_email_rate_limit( + _context(user_id), now=observed_at + ) + + same_scope = await asyncio.gather(*(attempt("user-1") for _ in range(11))) + other_scope = await asyncio.gather(*(attempt("user-2") for _ in range(10))) + other_workspace = await asyncio.gather( + *( + limiter_module.enforce_send_email_rate_limit( + _context("user-1", workspace_id="workspace-2"), now=observed_at + ) + for _ in range(10) + ) + ) + + assert sum(decision.allowed for decision in same_scope) == 10 + assert sum(not decision.allowed for decision in same_scope) == 1 + assert all(decision.allowed for decision in other_scope) + assert all(decision.allowed for decision in other_workspace) + + rollover = await limiter_module.enforce_send_email_rate_limit( + _context("user-1"), + now=observed_at + datetime.timedelta(seconds=61), + ) + assert rollover.allowed is True + scope_uid = ( + "email_send_scope:" + f"{limiter_module.rate_limit_scope_hash('user-1', 'org-1', 'workspace-org-1')}" + ) + assert sum(event.resource_uid == scope_uid for event in store.events) == 2 + + +@pytest.mark.asyncio +async def test_sliding_window_blocks_boundary_burst(monkeypatch): + store = _SharedAttemptStore() + monkeypatch.setattr( + limiter_module, "AsyncSessionLocal", lambda: _PostgresSession(store) + ) + started_at = datetime.datetime(2026, 8, 19, tzinfo=datetime.timezone.utc) + + for offset_seconds in range(50, 60): + decision = await limiter_module.enforce_send_email_rate_limit( + _context(), + now=started_at + datetime.timedelta(seconds=offset_seconds), + ) + assert decision.allowed is True + + blocked = await limiter_module.enforce_send_email_rate_limit( + _context(), + now=started_at + datetime.timedelta(seconds=61), + ) + assert blocked.allowed is False + for offset_seconds in range(62, 72): + assert not ( + await limiter_module.enforce_send_email_rate_limit( + _context(), + now=started_at + datetime.timedelta(seconds=offset_seconds), + ) + ).allowed + assert ( + sum( + event.event_action == "email_send_rate_limit.quota_exhausted" + for event in store.events + ) + == 1 + ) + + +@pytest.mark.asyncio +async def test_limiter_prunes_expired_allowed_state_but_keeps_denial_evidence( + monkeypatch, +): + store = _SharedAttemptStore() + observed_at = datetime.datetime(2026, 8, 19, 12, 0, tzinfo=datetime.timezone.utc) + scope_hash = limiter_module.rate_limit_scope_hash( + "user-1", "org-1", "workspace-org-1" + ) + scope_uid = f"email_send_scope:{scope_hash}" + expired_at = observed_at - datetime.timedelta(seconds=61) + expired_allowed = SecurityAuditEvent( + actor_user_id="user-1", + actor_role="member", + organization_id="org-1", + workspace_id="workspace-org-1", + event_action="email_send_rate_limit.allowed", + resource_type="email_send_rate_limit", + resource_uid=scope_uid, + evidence_source="services.email_send_rate_limiter", + observed_at=expired_at, + ) + durable_denial = SecurityAuditEvent( + actor_user_id="user-1", + actor_role="member", + organization_id="org-1", + workspace_id="workspace-org-1", + event_action="email_send_rate_limit.quota_exhausted", + resource_type="email_send_rate_limit", + resource_uid=scope_uid, + evidence_source="services.email_send_rate_limiter", + observed_at=expired_at, + ) + store.events.extend([expired_allowed, durable_denial]) + session = _PostgresSession(store) + monkeypatch.setattr(limiter_module, "AsyncSessionLocal", lambda: session) + + decision = await limiter_module.enforce_send_email_rate_limit( + _context(), now=observed_at + ) + + assert decision.allowed is True + assert expired_allowed not in store.events + assert durable_denial in store.events + assert ( + sum( + event.event_action == "email_send_rate_limit.allowed" + and event.resource_uid == scope_uid + for event in store.events + ) + == 1 + ) + delete_query_index = next( + index + for index, query in enumerate(session.queries) + if query.startswith("delete from security_audit_events") + ) + count_query_index = next( + index + for index, query in enumerate(session.queries) + if "count(" in query and "security_audit_events" in query + ) + assert delete_query_index < count_query_index + + +@pytest.mark.asyncio +async def test_limiter_uses_owned_transaction_and_non_sensitive_audit_state( + monkeypatch, +): + store = _SharedAttemptStore() + session = _PostgresSession(store) + monkeypatch.setattr(limiter_module, "AsyncSessionLocal", lambda: session) + decision = await limiter_module.enforce_send_email_rate_limit( + _context(), + now=datetime.datetime(2026, 8, 19, tzinfo=datetime.timezone.utc), + ) + + assert decision.allowed is True + assert any("pg_advisory_xact_lock" in query for query in session.queries) + assert session.isolation_level == "READ COMMITTED" + audit = next(item for item in session.added if isinstance(item, SecurityAuditEvent)) + assert "user-1" not in repr(audit) + assert "org-1" not in repr(audit) + assert audit.event_action == "email_send_rate_limit.allowed" + + +@pytest.mark.asyncio +async def test_limiter_uses_database_clock_after_scope_lock(monkeypatch): + store = _SharedAttemptStore() + database_now = datetime.datetime( + 2026, 8, 19, 12, 34, 56, tzinfo=datetime.timezone.utc + ) + session = _PostgresSession(store, database_now=database_now) + monkeypatch.setattr(limiter_module, "AsyncSessionLocal", lambda: session) + + decision = await limiter_module.enforce_send_email_rate_limit(_context()) + + assert decision.allowed is True + lock_query_index = next( + index + for index, query in enumerate(session.queries) + if "pg_advisory_xact_lock" in query + ) + clock_query_index = next( + index + for index, query in enumerate(session.queries) + if "clock_timestamp" in query + ) + assert lock_query_index < clock_query_index + audit = next(item for item in session.added if isinstance(item, SecurityAuditEvent)) + assert audit.observed_at == database_now + + +@pytest.mark.asyncio +async def test_rate_limiter_fails_closed_when_shared_state_is_unavailable(): + class _UnavailableSession: + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + def get_bind(self): + return SimpleNamespace(dialect=SimpleNamespace(name="sqlite")) + + original_factory = limiter_module.AsyncSessionLocal + limiter_module.AsyncSessionLocal = lambda: _UnavailableSession() + try: + with pytest.raises(limiter_module.EmailSendRateLimitUnavailable): + await limiter_module.enforce_send_email_rate_limit(_context()) + finally: + limiter_module.AsyncSessionLocal = original_factory + + +@pytest_asyncio.fixture +async def migrated_limiter_sessions(monkeypatch): + """Use the runner's migrated schema, never create missing runtime tables.""" + from asyncpg.exceptions import InvalidAuthorizationSpecificationError + from asyncpg.exceptions import InvalidPasswordError + from sqlalchemy.exc import OperationalError + + engine = create_async_engine(settings.DATABASE_URL, pool_size=4, max_overflow=0) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + scope_suffix = uuid.uuid4().hex + try: + try: + async with session_factory() as session: + await session.execute(select(SecurityAuditEvent.event_uid).limit(0)) + except ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, + OperationalError, + OSError, + ): + pytest.skip("PostgreSQL smoke database unavailable") + monkeypatch.setattr(limiter_module, "AsyncSessionLocal", session_factory) + try: + yield session_factory, scope_suffix + finally: + async with session_factory() as session: + await session.execute( + delete(SecurityAuditEvent).where( + SecurityAuditEvent.actor_user_id.in_( + [f"send-user-{scope_suffix}", f"send-peer-{scope_suffix}"] + ) + ) + ) + await session.commit() + finally: + await engine.dispose() + + +@pytest.mark.postgres +@pytest.mark.asyncio +async def test_real_postgres_concurrent_quota_isolates_each_scope_dimension( + migrated_limiter_sessions, +): + """Missing locks or a missing scope dimension must violate the 10/20 result.""" + session_factory, scope_suffix = migrated_limiter_sessions + contexts = [ + _context(f"send-user-{scope_suffix}", "send-org", "send-workspace"), + _context(f"send-peer-{scope_suffix}", "send-org", "send-workspace"), + _context(f"send-user-{scope_suffix}", "send-other-org", "send-workspace"), + _context(f"send-user-{scope_suffix}", "send-org", "send-other-workspace"), + ] + decisions = await asyncio.gather( + *( + limiter_module.enforce_send_email_rate_limit(context) + for context in contexts + for _ in range(20) + ) + ) + for scope_index, context in enumerate(contexts): + scope_decisions = decisions[scope_index * 20 : (scope_index + 1) * 20] + assert sum(decision.allowed for decision in scope_decisions) == 10 + async with session_factory() as session: + action_counts = dict( + ( + await session.execute( + select(SecurityAuditEvent.event_action, func.count()) + .where( + SecurityAuditEvent.actor_user_id == context.user_id, + SecurityAuditEvent.organization_id + == context.organization_id, + SecurityAuditEvent.workspace_id == context.workspace_id, + ) + .group_by(SecurityAuditEvent.event_action) + ) + ).all() + ) + assert action_counts == { + "email_send_rate_limit.allowed": 10, + "email_send_rate_limit.quota_exhausted": 1, + } + + +@pytest.mark.postgres +@pytest.mark.asyncio +async def test_real_postgres_window_expires_on_database_clock( + migrated_limiter_sessions, +): + """Wait the real production window; no reduced quota or injected timestamp.""" + session_factory, scope_suffix = migrated_limiter_sessions + context = _context(f"send-user-{scope_suffix}") + for _ in range(10): + assert (await limiter_module.enforce_send_email_rate_limit(context)).allowed + assert not (await limiter_module.enforce_send_email_rate_limit(context)).allowed + await asyncio.sleep(61) + assert (await limiter_module.enforce_send_email_rate_limit(context)).allowed + async with session_factory() as session: + actions = ( + ( + await session.execute( + select(SecurityAuditEvent.event_action).where( + SecurityAuditEvent.actor_user_id == context.user_id + ) + ) + ) + .scalars() + .all() + ) + assert sorted(actions) == [ + "email_send_rate_limit.allowed", + "email_send_rate_limit.quota_exhausted", + ] + + +@pytest.mark.postgres +@pytest.mark.asyncio +async def test_cancelled_lock_wait_returns_single_pool_slot( + migrated_limiter_sessions, + monkeypatch, +): + """A cancelled in-flight reservation must release its connection, not quota.""" + session_factory, scope_suffix = migrated_limiter_sessions + context = _context(f"send-user-{scope_suffix}") + engine = create_async_engine( + settings.DATABASE_URL, + pool_size=1, + max_overflow=0, + pool_timeout=1, + connect_args={"server_settings": {"application_name": scope_suffix}}, + ) + monkeypatch.setattr(limiter_module, "AsyncSessionLocal", async_sessionmaker(engine)) + pending_attempt = None + try: + async with session_factory() as lock_session: + scope_hash = limiter_module.rate_limit_scope_hash( + context.user_id, context.organization_id, context.workspace_id + ) + await lock_session.execute( + select(func.pg_advisory_xact_lock(limiter_module._lock_key(scope_hash))) + ) + pending_attempt = asyncio.create_task( + limiter_module.enforce_send_email_rate_limit(context) + ) + async with asyncio.timeout(5): + while True: + # Statistics are transaction-cached; refresh the observation + # without releasing the transaction-scoped barrier lock. + await lock_session.execute(select(func.pg_stat_clear_snapshot())) + waiting_count = ( + await lock_session.execute( + text( + "SELECT count(*) FROM pg_stat_activity " + "WHERE application_name = :application_name " + "AND wait_event = 'advisory'" + ), + {"application_name": scope_suffix}, + ) + ).scalar_one() + if waiting_count: + break + await asyncio.sleep(0.01) + pending_attempt.cancel() + with pytest.raises(asyncio.CancelledError): + await pending_attempt + assert engine.pool.checkedout() == 0 + await lock_session.rollback() + assert (await limiter_module.enforce_send_email_rate_limit(context)).allowed + async with session_factory() as session: + assert ( + await session.execute( + select(func.count()) + .select_from(SecurityAuditEvent) + .where(SecurityAuditEvent.actor_user_id == context.user_id) + ) + ).scalar_one() == 1 + finally: + if pending_attempt is not None: + pending_attempt.cancel() + await asyncio.gather(pending_attempt, return_exceptions=True) + await engine.dispose() + + +@pytest.mark.postgres +@pytest.mark.asyncio +async def test_signed_send_route_shares_single_pool_slot( + migrated_limiter_sessions, monkeypatch +): + """Dropping the request rollback must prevent this signed send from finishing.""" + _, scope_suffix = migrated_limiter_sessions + engine = create_async_engine( + settings.DATABASE_URL, + pool_size=1, + max_overflow=0, + pool_timeout=1, + ) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + user_id = f"send-user-{scope_suffix}" + + async def request_session(): + """Keep the real request session open until the HTTP request exits.""" + async with session_factory() as session: + yield session + + async def smtp_boundary(*, message_params, smtp_config): + """Do not send mail; verify only the call at the network boundary.""" + assert message_params.to_address == "recipient@example.com" + assert smtp_config.smtp_server == "mail.example.invalid" + assert smtp_config.smtp_username == user_id + assert engine.pool.checkedout() == 0 + return {"status": "sent", "simulated": False} + + monkeypatch.setitem(app.dependency_overrides, get_db, request_session) + monkeypatch.setattr(limiter_module, "AsyncSessionLocal", session_factory) + monkeypatch.setattr(emails_api, "send_email", smtp_boundary) + monkeypatch.setattr( + emails_api, + "validate_smtp_destination", + lambda smtp_server, smtp_port: (smtp_server, smtp_port), + ) + token = jwt.encode( + { + "ver": 1, + "iss": "naruon-control-plane", + "aud": "naruon-api", + "sub": user_id, + "role": "member", + "org": "send-org", + "groups": [], + "workspace": "send-workspace", + "exp": int(time.time()) + 300, + }, + settings.AUTH_SESSION_HMAC_SECRET.get_secret_value(), + algorithm="HS256", + ) + try: + async with session_factory() as session: + session.add( + TenantConfig( + user_id=user_id, + organization_id="send-org", + smtp_port=587, + smtp_server="mail.example.invalid", + smtp_username=user_id, + ) + ) + await session.commit() + async with AsyncClient( + transport=ASGITransport(app), base_url="http://test" + ) as client: + response = await client.post( + "/api/emails/send", + headers={"Authorization": f"Bearer {token}"}, + json={ + "to": "recipient@example.com", + "subject": "Pool regression", + "body": "Body", + }, + ) + assert response.status_code == 200, response.json() + assert response.json() == {"status": "sent", "simulated": False} + assert engine.pool.checkedout() == 0 + async with session_factory() as session: + assert ( + await session.execute( + select(func.count()) + .select_from(SecurityAuditEvent) + .where(SecurityAuditEvent.actor_user_id == user_id) + ) + ).scalar_one() == 1 + finally: + try: + async with session_factory() as session: + await session.execute( + delete(TenantConfig).where(TenantConfig.user_id == user_id) + ) + await session.commit() + finally: + await engine.dispose() diff --git a/backend/tests/test_emails_api.py b/backend/tests/test_emails_api.py index 7bffa6ff7..3c06363b7 100644 --- a/backend/tests/test_emails_api.py +++ b/backend/tests/test_emails_api.py @@ -23,6 +23,10 @@ from unittest.mock import AsyncMock, patch from services.embedding import STORAGE_EMBEDDING_DIMENSION from services.email_service import generate_email_fingerprint +from services.email_send_rate_limiter import ( + EmailSendRateLimitDecision, + EmailSendRateLimitUnavailable, +) pytestmark = pytest.mark.usefixtures("dev_auth_dependency_overrides") TEST_SESSION_HMAC_SECRET = os.environ["AUTH_SESSION_HMAC_SECRET"] @@ -1782,6 +1786,11 @@ def fake_validate_smtp_destination(smtp_server, smtp_port, *, resolve_host=True) "api.emails.validate_smtp_destination", fake_validate_smtp_destination ) + async def allow_rate_limit(*args, **kwargs): + return EmailSendRateLimitDecision(allowed=True, reason="allowed") + + monkeypatch.setattr(emails_api, "enforce_send_email_rate_limit", allow_rate_limit) + client = TestClient(app, headers={"X-User-Id": "testuser"}) response = client.post( @@ -1864,9 +1873,18 @@ def fake_validate_smtp_destination(smtp_server, smtp_port, *, resolve_host=True) monkeypatch.setattr( "api.emails.validate_smtp_destination", fake_validate_smtp_destination ) - monkeypatch.setattr(emails_api, "_SEND_EMAIL_RATE_LIMIT_MAX_ATTEMPTS", 1) - monkeypatch.setattr(emails_api.time, "monotonic", lambda: 100.0) - emails_api._email_send_attempts_by_scope.clear() + + calls = 0 + + async def fake_rate_limit(*args, **kwargs): + nonlocal calls + calls += 1 + return EmailSendRateLimitDecision( + allowed=calls == 1, + reason="allowed" if calls == 1 else "quota_exhausted", + ) + + monkeypatch.setattr(emails_api, "enforce_send_email_rate_limit", fake_rate_limit) try: client = TestClient(app, headers={"X-User-Id": "testuser"}) @@ -1879,13 +1897,42 @@ def fake_validate_smtp_destination(smtp_server, smtp_port, *, resolve_host=True) assert client.post("/api/emails/send", json=payload).status_code == 200 response = client.post("/api/emails/send", json=payload) finally: - emails_api._email_send_attempts_by_scope.clear() + app.dependency_overrides.clear() assert response.status_code == 429 assert response.json() == {"detail": "Email send rate limit exceeded"} mock_send_email.assert_called_once() +@patch("api.emails.send_email", return_value={"status": "sent", "simulated": False}) +def test_send_email_endpoint_fails_closed_when_rate_limiter_is_unavailable( + mock_send_email, monkeypatch +): + from fastapi.testclient import TestClient + from main import app + + monkeypatch.setattr( + "api.emails.validate_smtp_destination", + lambda *args, **kwargs: (args[0], args[1]), + ) + + async def unavailable_rate_limit(*args, **kwargs): + raise EmailSendRateLimitUnavailable + + monkeypatch.setattr( + emails_api, "enforce_send_email_rate_limit", unavailable_rate_limit + ) + + response = TestClient(app, headers={"X-User-Id": "testuser"}).post( + "/api/emails/send", + json={"to": "test@example.com", "subject": "Test", "body": "Body"}, + ) + + assert response.status_code == 503 + assert response.json() == {"detail": "Email send rate limiter unavailable"} + mock_send_email.assert_not_called() + + @patch("api.emails.send_email", return_value={"status": "simulated", "simulated": True}) def test_send_email_endpoint_ignores_user_id_query_and_uses_authenticated_user_config( mock_send_email, monkeypatch, sample_email @@ -1900,6 +1947,10 @@ def fake_validate_smtp_destination(smtp_server, smtp_port, *, resolve_host=True) monkeypatch.setattr( "api.emails.validate_smtp_destination", fake_validate_smtp_destination ) + async def allow_rate_limit(*args, **kwargs): + return EmailSendRateLimitDecision(allowed=True, reason="allowed") + + monkeypatch.setattr(emails_api, "enforce_send_email_rate_limit", allow_rate_limit) session = ScalarQueryCapturingSession([sample_email]) async def tenant_db(): @@ -2006,6 +2057,10 @@ def fake_validate_smtp_destination(smtp_server, smtp_port, *, resolve_host=True) monkeypatch.setattr( "api.emails.validate_smtp_destination", fake_validate_smtp_destination ) + async def allow_rate_limit(*args, **kwargs): + return EmailSendRateLimitDecision(allowed=True, reason="allowed") + + monkeypatch.setattr(emails_api, "enforce_send_email_rate_limit", allow_rate_limit) client = TestClient(app, headers={"X-User-Id": "testuser"}) diff --git a/backend/tests/test_security_audit_migration_contract.py b/backend/tests/test_security_audit_migration_contract.py new file mode 100644 index 000000000..1549f9142 --- /dev/null +++ b/backend/tests/test_security_audit_migration_contract.py @@ -0,0 +1,41 @@ +"""Regression contract for durable security-audit schema migration.""" + +from pathlib import Path + + +BACKEND_ROOT = Path(__file__).resolve().parents[1] +MIGRATION_PATH = ( + BACKEND_ROOT / "alembic" / "versions" / "0018_security_audit_events.py" +) + + +def test_security_audit_events_have_structured_incremental_revision() -> None: + """Alembic upgrades must create the durable audit table used by runtime gates.""" + assert MIGRATION_PATH.exists() + revision_text = MIGRATION_PATH.read_text(encoding="utf-8") + + assert 'revision = "0018_security_audit_events"' in revision_text + assert 'down_revision = "0017_merge_newsdom_carddav_heads"' in revision_text + assert '"security_audit_events"' in revision_text + for column_name in ( + "event_uid", + "actor_user_id", + "actor_role", + "organization_id", + "workspace_id", + "event_action", + "resource_type", + "resource_uid", + "evidence_source", + "detail_text", + "observed_at", + ): + assert f'"{column_name}"' in revision_text + + assert "op.create_table(" in revision_text + assert "has_table" in revision_text + assert "ix_security_audit_events_scope_time" in revision_text + assert "ix_security_audit_events_actor_scope" in revision_text + assert "op.create_index(" in revision_text + assert "if_not_exists=True" in revision_text + assert "Base.metadata.create_all" not in revision_text diff --git a/docs/doctoring/shared_send_postgres.md b/docs/doctoring/shared_send_postgres.md new file mode 100644 index 000000000..f78820635 --- /dev/null +++ b/docs/doctoring/shared_send_postgres.md @@ -0,0 +1,128 @@ +# Shared-send migrated PostgreSQL evidence + +Status: Proposed; existing source owner [PR #1417](https://github.com/ContextualWisdomLab/naruon/pull/1417). +Local checks do not establish protected integration, delivery, or performance. + +## Problem, dependency and preserved histories + +Starting head `a9f334a442538b666e03e694731745d8aab4b45a` uses a PostgreSQL +transaction-scoped advisory lock, database clock, rolling 10-attempt/60-second +quota, and user/organization/workspace scope. However, its database smoke +creates ORM tables, reduces the quota to one and injects time. Its one-slot +pool regression uses a semaphore double. Neither proves the migrated runtime +or real pool/cancellation behavior. + +Before committing, claim the existing owner and normally merge complete CI +prerequisite #1562 at `4d2e4abc2c369d5e85bced4027b6f81857721ea2`, including +#1503/#1565/#1571, rather than copy a runner or migration. This produced a real +fresh-install RED: Alembic reported two heads, `0018_security_audit_events` +and `0019_email_read_state_repair`. Preserve both histories with +`0020_merge_send_registry`, a no-DDL merge revision. Do not delete either parent, +stamp over missing work, switch the deployment target to a partial branch, or +claim an unreleased foundation is available. Only CHANGELOG had a textual merge +conflict; both entries were retained. Keep this PR Draft on the CI prerequisite. + +Alembic's merge-node mechanism makes both parent paths prerequisites of one +head (Bayer, n.d.). The same runner now completes fresh and repeated upgrades. +This is repository-local stack integration, not permission to consume source +from an external service or bypass an immutable release contract. + +## Behavioral checks and failure analysis + +The updated database fixture reads the migrated audit table without creating +it. New tests use the production quota and database time: 80 concurrent attempts +across four scopes produce 10 allowed reservations and one durable quota event +per scope; the real 61-second wait verifies expiration without accelerating the +clock. Each dimension changes independently, so a missing user, organization or +workspace predicate cannot borrow another dimension's isolation. + +A separate one-slot engine queues behind a real advisory lock. The test waits +until PostgreSQL observes that wait, cancels the task, verifies no checked-out +connection remains and confirms a subsequent attempt creates exactly one +reservation. Initial observation failed because PostgreSQL caches activity +snapshots within a transaction. Refreshing with `pg_stat_clear_snapshot()` fixes +the test observation without releasing the barrier lock; this was not a runtime +limiter defect (PostgreSQL Global Development Group, n.d.-b). + +A real signed backend HTTP request reads persisted tenant configuration and +uses the same one-slot pool as the real limiter. Only DNS validation and SMTP +delivery are substituted at the external boundary. Removing the existing +request rollback temporarily makes this test fail with HTTP 503 and exhausted +pool capacity; restoring it recovers the existing behavior. No mutation is +retained. An initially invalid reserved recipient suffix was corrected in the +test fixture; validation was not weakened. This does not exercise the browser +cookie/proxy path, prove actual SMTP delivery, or measure customer latency. + +Before final commit, the focused four-file run passed 74 tests in 62.09 seconds +with fresh/repeated migrations and completed test-only container/network cleanup. +The final PR receipt must name the unchanged committed head and JUnit digest. +Subsequent whole-suite execution must first inherit #1562's nested subprocess +isolation follow-up; the parent environment alone does not protect children +that replace it. Retain failed candidates separately from GREEN evidence. + +Run after installing the existing hash-locked core and Noema requirements: + +```sh +bash scripts/ci/run_backend_postgres.sh +``` + +Ponytail reuse keeps the existing limiter, transaction cleanup and CI lifecycle; +only migration reconciliation and missing behavior checks are added. No new +runtime package, generic abstraction, quota heuristic or model timeout is needed. +Remaining acceptance: current-head hosted checks and independent review, parent +integration, protected merge, delivery/browser evidence, and realistic load. +No coverage percentage, release, deployment or p95 claim is established here. + +## SMTP connection cancellation follow-up + +At `1666f76cf94c31e34c2762c9d75f52ea3040b9a2`, the shared +`services.email_client._connect_validated_smtp_socket` closes the raw socket for +`Exception`, but `asyncio.CancelledError` derives from `BaseException` (Python +Software Foundation, n.d.). Cancellation while awaiting `sock_connect` therefore +escapes before the socket reaches the caller's `finally`. A cancelled send can +retain an open descriptor until later object reclamation. Both direct +`send_email` and `runner.local_mail_adapters.LocalMailAdapters.send_smtp` use +this helper; the latter is not a separate SMTP implementation. + +The two new `test_smtp_connect_cancellation_closes_socket` cases exercise those +real consumers and create real OS sockets. Only DNS and the pending external +connect are substituted. After the connection starts, cancelling the actual +task must propagate the cancellation message and leave `fileno() == -1` before +the test releases its socket reference. Both cases were RED with an open +descriptor (`14 != -1`); the test's own `finally` still closes it so a failing +candidate does not leak the test resource. A five-second synchronization guard +bounds the test observation, not a model or application timeout. + +Extend the existing cleanup handler to include `asyncio.CancelledError`, close +the socket synchronously, and re-raise without replacement or suppression. +Per-caller guards were rejected because neither caller owns the socket before +the helper returns. A new resource abstraction or dependency adds no value to +this one-line ownership fix. DNS pinning, TLS hostname checks, quota semantics, +and successful ownership transfer remain unchanged. The focused SMTP, message, +and registered-adapter suites passed 55 tests with warnings treated as errors. +The final exact-head whole-suite receipt belongs in PR #1417. + +DeepWiki's September 6 lookup described a nonexistent connection `finally` and +excluded the registered Connector from the shared path. Current CodeGraph +source/caller evidence and the RED reproduction contradict that summary; do +not treat generated documentation as exact-revision runtime evidence. This +follow-up proves deterministic cleanup before connection completion, not real +SMTP delivery, browser cancellation propagation, recipient acceptance, or +post-delivery retry safety. No external message was sent. + +## References (APA 7) + +Bayer, M. (n.d.). *Working with branches*. Alembic documentation. Retrieved +September 6, 2026, from https://alembic.sqlalchemy.org/en/latest/branches.html + +PostgreSQL Global Development Group. (n.d.-a). *Explicit locking*. +PostgreSQL 16 documentation. Retrieved September 6, 2026, from +https://www.postgresql.org/docs/16/explicit-locking.html#ADVISORY-LOCKS + +PostgreSQL Global Development Group. (n.d.-b). *The cumulative statistics system*. +PostgreSQL 16 documentation. Retrieved September 6, 2026, from +https://www.postgresql.org/docs/16/monitoring-stats.html + +Python Software Foundation. (n.d.). *Exceptions*. Python 3.14 documentation. +Retrieved September 6, 2026, from +https://docs.python.org/3.14/library/asyncio-exceptions.html