mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-13 00:14:02 +00:00
arc53-machine
6b4220d401
- An MCP connection's tokens only go to its own server, and a
connection_id can no longer come from a client: the MCP test/save
routes drop it and the tool executor uses only the resolved one.
- API-key connections are reused only when they hold the same
credentials. Two keys that share a hint or a label get separate
connections ("…abcd (2)"), at runtime and in the 0038 backfill.
- Reconnecting a connection whose secrets cannot be decrypted no longer
flags it from a second transaction while holding its row lock.
- Connection setup validates the sync request before claiming its
Idempotency-Key, so a corrected retry is queued.
- A request that names no connection resolves to none without opening
a database connection; fixes two tests that reached a real database.
105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
"""Regression tests: connector auth must not leak session tokens.
|
|
|
|
Three connector auth modules previously interpolated the raw session
|
|
token into ``ValueError`` messages. With ``exc_info=True`` on upstream
|
|
loggers those messages land in ``stack_logs`` (Postgres) and Sentry.
|
|
These tests pin the behaviour so the raw token never reappears in the
|
|
raised exception's ``str()`` representation, while a stable, short
|
|
SHA-256 fingerprint is present for correlation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, Optional
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from docsgpt.parser.connectors._auth_utils import session_token_fingerprint
|
|
|
|
|
|
SECRET_TOKEN = "super-secret-session-token-ABCDEF1234567890"
|
|
|
|
|
|
class _FakeRepo:
|
|
"""Fake ``ConnectorSessionsRepository`` returning a preset session."""
|
|
|
|
_session: Optional[Dict[str, Any]] = None
|
|
|
|
def __init__(self, conn: Any) -> None:
|
|
self.conn = conn
|
|
|
|
def get_by_session_token(self, session_token: str) -> Optional[Dict[str, Any]]:
|
|
return self._session
|
|
|
|
|
|
class _FakeReadonlyCtx:
|
|
"""Fake ``db_readonly`` context manager yielding a dummy connection."""
|
|
|
|
def __enter__(self) -> MagicMock:
|
|
return MagicMock()
|
|
|
|
def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> bool:
|
|
return False
|
|
|
|
|
|
def _patches(session_return: Optional[Dict[str, Any]]):
|
|
fake_repo_cls = type(
|
|
"FakeRepo",
|
|
(_FakeRepo,),
|
|
{"_session": session_return},
|
|
)
|
|
return (
|
|
patch(
|
|
"docsgpt.connectors.service.ConnectorSessionsRepository",
|
|
fake_repo_cls,
|
|
),
|
|
patch(
|
|
"docsgpt.connectors.service.db_readonly",
|
|
lambda: _FakeReadonlyCtx(),
|
|
),
|
|
)
|
|
|
|
|
|
class TestSessionTokenFingerprint:
|
|
"""Unit tests for the shared fingerprint helper."""
|
|
|
|
@pytest.mark.unit
|
|
def test_fingerprint_is_stable(self) -> None:
|
|
assert session_token_fingerprint("abc") == session_token_fingerprint("abc")
|
|
|
|
@pytest.mark.unit
|
|
def test_fingerprint_does_not_contain_token(self) -> None:
|
|
fp = session_token_fingerprint(SECRET_TOKEN)
|
|
assert SECRET_TOKEN not in fp
|
|
assert fp.startswith("sha256:")
|
|
# 6 hex chars after the prefix.
|
|
assert len(fp) == len("sha256:") + 6
|
|
|
|
@pytest.mark.unit
|
|
def test_empty_token_has_sentinel(self) -> None:
|
|
assert session_token_fingerprint("") == "sha256:<empty>"
|
|
# type-check intentionally ignored: defensive for None.
|
|
assert session_token_fingerprint(None) == "sha256:<empty>" # type: ignore[arg-type]
|
|
|
|
@pytest.mark.unit
|
|
def test_different_tokens_produce_different_fingerprints(self) -> None:
|
|
assert session_token_fingerprint("a") != session_token_fingerprint("b")
|
|
|
|
|
|
class TestConnectionLookupDoesNotLeakToken:
|
|
"""A session token that names no connection is reported by fingerprint only."""
|
|
|
|
@pytest.mark.unit
|
|
def test_invalid_session_does_not_interpolate_token(self) -> None:
|
|
from docsgpt.connectors import service
|
|
|
|
repo_patch, ctx_patch = _patches(None)
|
|
with repo_patch, ctx_patch:
|
|
with pytest.raises(ValueError) as excinfo:
|
|
service.connection_id_for_session_token(SECRET_TOKEN)
|
|
|
|
message = str(excinfo.value)
|
|
assert SECRET_TOKEN not in message
|
|
assert session_token_fingerprint(SECRET_TOKEN) in message
|