Agentul tocmai a primit acces la calendarul firmei. Înainte să ajungă lângă date reale, trebuie demonstrat că poate citi cu un token limitat, că nu poate scrie fără permisiunea potrivită, că revocarea îi oprește accesul și că un timeout nu produce o acțiune repetată sau ambiguă.
La finalul acestui laborator rulează un conector local FastAPI, un client asincron asemănător componentei care ar executa tool-urile unui agent și opt teste automate. Suita verifică autorizarea pe scope-uri, tokenurile necunoscute sau revocate, jurnalul de audit înlănțuit cu HMAC, absența tokenurilor din loguri și oprirea controlată a unei operații când conectorul nu răspunde.
Nu conectăm un calendar real și nu folosim un LLM. Astfel, rezultatul măsoară contractul de securitate al conectorului, nu comportamentul probabilistic al modelului. Pentru evaluarea separată a alegerii tool-urilor poți continua cu ghidul despre evaluări pentru agenți AI cu tool-uri.
Fixăm contractul la un calendar izolat și două drepturi distincte
Laboratorul folosește următorul contract:
calendar.readpermite doar citirea evenimentelor;calendar.writepermite crearea unui eveniment;- tokenul de scriere primește ambele scope-uri, iar cel de citire numai
calendar.read; - un token necunoscut sau revocat primește HTTP
401; - un token valid, dar insuficient, primește HTTP
403; - fiecare apel este înregistrat fără valoarea tokenului;
- o eroare de transport oprește apelul și marchează rezultatul drept neconfirmat;
- operația de scriere nu este reluată automat.
Răspunsurile 401 și 403 includ antetul WWW-Authenticate, folosind erorile invalid_token și insufficient_scope descrise de RFC 6750. Revocarea este simulată în registrul local al laboratorului. Într-o integrare reală, ea aparține serverului de autorizare sau furnizorului de identitate; RFC 7009 definește un endpoint OAuth pentru revocarea tokenurilor.
Această separare este intenționat mai îngustă decât o integrare MCP completă. Dacă agentul folosește MCP, modelul de acces poate fi extins cu OAuth 2.1 și permisiuni granulare pentru tool-uri. Testele de aici rămân utile la granița dintre executorul agentului și serviciul extern.
Folosim versiuni fixate și verificăm mașina înainte de instalare
Configurația de referință este Ubuntu Server 24.04 LTS, Python 3.12 și un proces local fără GPU. Proiectul nu antrenează un model și nu are nevoie de accelerator. Pentru laborator sunt suficiente un nucleu CPU, aproximativ 1 GB RAM disponibil și 200 MB liberi pe disc. Durata estimată este de 30–45 de minute.
Rulează mai întâi comenzile de diagnostic:
cat /etc/os-release
python3 --version
uname -m
free -h
df -h "$HOME"Un rezultat compatibil arată astfel; versiunea minoră a sistemului, arhitectura și valorile de memorie pot diferi:
NAME="Ubuntu"
VERSION="24.04.x LTS (Noble Numbat)"
Python 3.12.x
x86_64
Mem: ... cel puțin 1 GiB disponibil pentru laborator
... cel puțin 200 MiB disponibili în /homeBibliotecile sunt blocate la FastAPI 0.115.6, HTTPX 0.28.1, pytest 8.3.4, pytest-asyncio 0.25.0 și Uvicorn 0.34.0. Nu sunt prezentate drept cele mai noi versiuni, ci drept setul reproductibil folosit în acest proiect.
Dacă directorul există deja, comanda se oprește pentru a nu suprascrie fișiere. Mută sau arhivează manual proiectul anterior înainte să continui.
set -euo pipefail
PROJECT="$HOME/connector-contract-lab"
if [ -e "$PROJECT" ]; then
printf 'Directorul există deja: %sn' "$PROJECT" >&2
exit 1
fi
sudo apt-get update
sudo apt-get install -y python3-venv
mkdir -p "$PROJECT"/{src/connector_lab,tests,scripts,artifacts}
cd "$PROJECT"
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pipPunct de control: promptul trebuie să indice mediul .venv, iar python --version trebuie să raporteze Python 3.12.x.
Harta laboratorului separă serviciul, executorul și probele
Proiectul va avea următoarea structură:
connector-contract-lab/
├── requirements.txt
├── src/
│ └── connector_lab/
│ ├── api.py # conectorul calendar și registrul de tokenuri
│ ├── audit.py # jurnal JSONL înlănțuit și verificarea integrității
│ └── client.py # executorul controlat folosit de agent
├── tests/
│ └── test_contract.py # cele opt teste de securitate și reziliență
├── scripts/
│ └── run_suite.sh # instalare verificată, testare și raport JUnit
└── artifacts/
├── audit.jsonl # evenimente fără tokenurile brute
└── junit.xml # raportul automat al suiteiRegistrul de tokenuri și evenimentele sunt ținute în memorie. Este o alegere pentru laborator: fiecare test pornește cu stare cunoscută și nu atinge servicii sau date reale.
Pasul 1 — Blocăm dependențele care definesc contractul testat
Fișierul requirements.txt:
fastapi==0.115.6
httpx==0.28.1
pytest==8.3.4
pytest-asyncio==0.25.0
uvicorn==0.34.0Instalează exact setul declarat:
python -m pip install --requirement requirements.txt
python -m pip checkpip check verifică dacă dependențele instalate declară incompatibilități între ele. Nu testează aplicația; această verificare vine mai târziu.
Exemplu de rezultat:
No broken requirements found.Punct de control: dacă pip check raportează un conflict, verifică dacă mediul virtual este activ și dacă instalarea s-a făcut în proiectul nou, nu într-un mediu Python reutilizat.
Pasul 2 — Construim conectorul cu scope-uri și revocare imediată
Fișierul src/connector_lab/api.py:
from dataclasses import dataclass
from hashlib import sha256
from threading import RLock
from fastapi import Depends, FastAPI, Header, HTTPException, status
from pydantic import BaseModel
@dataclass(frozen=True)
class TokenRecord:
subject: str
scopes: frozenset[str]
active: bool = True
class TokenStore:
def __init__(self) -> None:
self._records: dict[str, TokenRecord] = {}
self._lock = RLock()
@staticmethod
def _digest(token: str) -> str:
return sha256(token.encode("utf-8")).hexdigest()
def register(self, token: str, subject: str, scopes: set[str]) -> None:
record = TokenRecord(
subject=subject,
scopes=frozenset(scopes),
active=True,
)
with self._lock:
self._records[self._digest(token)] = record
def resolve(self, token: str) -> TokenRecord | None:
with self._lock:
return self._records.get(self._digest(token))
def revoke(self, token: str) -> bool:
digest = self._digest(token)
with self._lock:
record = self._records.get(digest)
if record is None:
return False
self._records[digest] = TokenRecord(
subject=record.subject,
scopes=record.scopes,
active=False,
)
return True
class EventCreate(BaseModel):
title: str
def create_app(store: TokenStore) -> FastAPI:
app = FastAPI(title="Calendar connector laboratory")
events: list[dict[str, str]] = [
{"id": "event-001", "title": "Revizuire acces agent"}
]
def require_scope(required_scope: str):
def authorize(
authorization: str | None = Header(default=None),
) -> TokenRecord:
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="missing or invalid bearer token",
headers={
"WWW-Authenticate": 'Bearer error="invalid_token"'
},
)
token = authorization.removeprefix("Bearer ").strip()
record = store.resolve(token)
if record is None or not record.active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="token is unknown or revoked",
headers={
"WWW-Authenticate": 'Bearer error="invalid_token"'
},
)
if required_scope not in record.scopes:
challenge = (
'Bearer error="insufficient_scope", '
f'scope="{required_scope}"'
)
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"required scope: {required_scope}",
headers={"WWW-Authenticate": challenge},
)
return record
return authorize
@app.get("/calendar/events")
def list_events(
_: TokenRecord = Depends(require_scope("calendar.read")),
) -> list[dict[str, str]]:
return list(events)
@app.post("/calendar/events", status_code=status.HTTP_201_CREATED)
def create_event(
event: EventCreate,
_: TokenRecord = Depends(require_scope("calendar.write")),
) -> dict[str, str]:
created = {
"id": f"event-{len(events) + 1:03d}",
"title": event.title,
}
events.append(created)
return created
return appTokenStore păstrează drept cheie doar SHA-256 al tokenului, nu valoarea brută. Acest lucru limitează expunerea accidentală în inspecții, dar nu transformă tokenurile slabe în secrete sigure: în producție, ele trebuie emise aleatoriu și gestionate de un sistem de autorizare.
frozenset împiedică modificarea accidentală a scope-urilor unui TokenRecord. Revocarea înlocuiește înregistrarea cu una inactivă și produce efect la următoarea cerere.
Rezultatul nu este încă un serviciu pornit; verifică doar că modulul se importă:
PYTHONPATH=src python -c
'from connector_lab.api import TokenStore, create_app; print(create_app(TokenStore()).title)'Exemplu de output:
Calendar connector laboratoryPunct de control: dacă importul eșuează, verifică existența căii src/connector_lab/api.py și valoarea PYTHONPATH=src.
Pasul 3 — Semnăm fiecare intrare din audit fără să salvăm credentialul
Fișierul src/connector_lab/audit.py:
import hashlib
import hmac
import json
from datetime import datetime, timezone
from pathlib import Path
from threading import Lock
from typing import Any
class AuditTrail:
def __init__(self, path: Path, key: bytes) -> None:
if len(key) < 32:
raise ValueError("audit key must contain at least 32 bytes")
self.path = path
self.key = key
self._lock = Lock()
self._sequence = 0
self._previous_hash = "0" * 64
self.path.parent.mkdir(parents=True, exist_ok=True)
@staticmethod
def token_fingerprint(token: str) -> str:
digest = hashlib.sha256(token.encode("utf-8")).hexdigest()
return digest[:12]
@staticmethod
def _canonical(entry: dict[str, Any]) -> bytes:
return json.dumps(
entry,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
def append(
self,
*,
action: str,
actor: str,
token: str,
outcome: str,
status_code: int,
) -> None:
with self._lock:
self._sequence += 1
entry: dict[str, Any] = {
"sequence": self._sequence,
"timestamp": datetime.now(timezone.utc).isoformat(),
"action": action,
"actor": actor,
"token_fingerprint": self.token_fingerprint(token),
"outcome": outcome,
"status_code": status_code,
"previous_hash": self._previous_hash,
}
entry_hash = hmac.new(
self.key,
self._canonical(entry),
hashlib.sha256,
).hexdigest()
entry["entry_hash"] = entry_hash
with self.path.open("a", encoding="utf-8") as handle:
handle.write(
json.dumps(entry, ensure_ascii=False, sort_keys=True)
+ "n"
)
self._previous_hash = entry_hash
def verify(self) -> int:
expected_previous = "0" * 64
expected_sequence = 1
with self.path.open("r", encoding="utf-8") as handle:
for line in handle:
entry = json.loads(line)
stored_hash = entry.pop("entry_hash")
if (
entry["sequence"] != expected_sequence
or entry["previous_hash"] != expected_previous
):
raise ValueError(
f"audit chain broken at sequence {expected_sequence}"
)
calculated_hash = hmac.new(
self.key,
self._canonical(entry),
hashlib.sha256,
).hexdigest()
if not hmac.compare_digest(stored_hash, calculated_hash):
raise ValueError(
f"audit chain broken at sequence {expected_sequence}"
)
expected_previous = stored_hash
expected_sequence += 1
return expected_sequence - 1Fiecare intrare include hash-ul celei anterioare și un HMAC-SHA-256 calculat cu o cheie de cel puțin 32 de octeți. hmac.compare_digest evită o comparație obișnuită a semnăturilor.
Jurnalul nu este imuabil în fața unei persoane care poate modifica atât fișierul, cât și cheia. În producție, cheia trebuie separată de aplicație, iar evenimentele expediate către un sistem central cu acces restrictiv și retenție controlată.
Verifică sintaxa:
python -m py_compile src/connector_lab/audit.pyLipsa output-ului indică un fișier valid sintactic.
Punct de control: mesajul audit key must contain at least 32 bytes înseamnă că testul sau aplicația a furnizat o cheie prea scurtă, nu că trebuie eliminată validarea.
Pasul 4 — Oprim executorul când rezultatul scrierii devine incert
Fișierul src/connector_lab/client.py:
import httpx
from connector_lab.audit import AuditTrail
class ConnectorRejected(RuntimeError):
def __init__(
self,
status_code: int,
detail: str,
challenge: str | None,
) -> None:
super().__init__(
f"connector rejected request with status {status_code}: {detail}"
)
self.status_code = status_code
self.detail = detail
self.challenge = challenge
class ConnectorUnavailable(RuntimeError):
pass
class ConnectorClient:
def __init__(
self,
*,
base_url: str,
token: str,
actor: str,
audit: AuditTrail,
transport: httpx.AsyncBaseTransport,
timeout_seconds: float = 0.2,
) -> None:
self.base_url = base_url
self.token = token
self.actor = actor
self.audit = audit
self.transport = transport
self.timeout = httpx.Timeout(timeout_seconds)
async def _request(
self,
method: str,
path: str,
*,
payload: dict[str, str] | None = None,
) -> httpx.Response:
action = f"{method} {path}"
try:
async with httpx.AsyncClient(
base_url=self.base_url,
transport=self.transport,
timeout=self.timeout,
) as client:
response = await client.request(
method,
path,
headers={"Authorization": f"Bearer {self.token}"},
json=payload,
)
except httpx.TransportError as exc:
self.audit.append(
action=action,
actor=self.actor,
token=self.token,
outcome="transport_error",
status_code=0,
)
raise ConnectorUnavailable(
"connector unavailable; action not confirmed"
) from exc
if response.status_code < 400:
outcome = "success"
elif response.status_code in {401, 403}:
outcome = "denied"
else:
outcome = "error"
self.audit.append(
action=action,
actor=self.actor,
token=self.token,
outcome=outcome,
status_code=response.status_code,
)
if response.status_code >= 400:
try:
detail = response.json().get("detail", response.text)
except ValueError:
detail = response.text
raise ConnectorRejected(
response.status_code,
str(detail),
response.headers.get("WWW-Authenticate"),
)
return response
async def list_events(self) -> list[dict[str, str]]:
response = await self._request("GET", "/calendar/events")
return response.json()
async def create_event(self, title: str) -> dict[str, str]:
response = await self._request(
"POST",
"/calendar/events",
payload={"title": title},
)
return response.json()Clientul acceptă un transport HTTPX injectat. Testele folosesc ASGITransport pentru a apela aplicația local, fără port de rețea, și MockTransport pentru a produce controlat o eroare.
Nu există retry automat. Pentru o citire, repetarea poate fi adăugată ulterior cu o politică explicită. Pentru o scriere, repetarea oarbă după timeout poate dubla acțiunea dacă serviciul a procesat cererea, dar răspunsul s-a pierdut. Mesajul action not confirmed obligă orchestratorul să oprească fluxul sau să verifice starea înaintea unei noi încercări. Aceeași limită trebuie aplicată stratului de planificare, așa cum este descris în ghidul despre separarea planificării de execuția agenților.
Verifică ambele module:
python -m py_compile
src/connector_lab/api.py
src/connector_lab/client.pyPunct de control: lipsa output-ului înseamnă că importurile și sintaxa sunt acceptate; nu dovedește încă aplicarea corectă a permisiunilor.
Pasul 5 — Executăm aceleași probe pentru acces permis, refuz și indisponibilitate
Fișierul tests/test_contract.py:
import os
from pathlib import Path
import httpx
import pytest
from connector_lab.api import TokenStore, create_app
from connector_lab.audit import AuditTrail
from connector_lab.client import (
ConnectorClient,
ConnectorRejected,
ConnectorUnavailable,
)
@pytest.fixture(scope="session")
def audit() -> AuditTrail:
path = Path("artifacts/audit.jsonl")
path.unlink(missing_ok=True)
trail = AuditTrail(path, os.urandom(32))
yield trail
assert trail.verify() > 0
@pytest.fixture
def connector_state():
store = TokenStore()
read_token = "read-" + os.urandom(24).hex()
write_token = "write-" + os.urandom(24).hex()
store.register(
read_token,
"agent-reader",
{"calendar.read"},
)
store.register(
write_token,
"agent-writer",
{"calendar.read", "calendar.write"},
)
return {
"store": store,
"app": create_app(store),
"read_token": read_token,
"write_token": write_token,
}
def make_client(
*,
token: str,
audit: AuditTrail,
transport: httpx.AsyncBaseTransport,
) -> ConnectorClient:
return ConnectorClient(
base_url="http://connector.test",
token=token,
actor="support-agent",
audit=audit,
transport=transport,
)
@pytest.mark.asyncio
async def test_read_scope_lists_events(audit, connector_state):
transport = httpx.ASGITransport(app=connector_state["app"])
client = make_client(
token=connector_state["read_token"],
audit=audit,
transport=transport,
)
events = await client.list_events()
assert events == [
{"id": "event-001", "title": "Revizuire acces agent"}
]
@pytest.mark.asyncio
async def test_read_scope_cannot_create_event(audit, connector_state):
transport = httpx.ASGITransport(app=connector_state["app"])
client = make_client(
token=connector_state["read_token"],
audit=audit,
transport=transport,
)
with pytest.raises(ConnectorRejected) as captured:
await client.create_event("Acțiune care trebuie refuzată")
assert captured.value.status_code == 403
assert captured.value.detail == "required scope: calendar.write"
assert "insufficient_scope" in (captured.value.challenge or "")
assert 'scope="calendar.write"' in (captured.value.challenge or "")
@pytest.mark.asyncio
async def test_write_scope_creates_event(audit, connector_state):
transport = httpx.ASGITransport(app=connector_state["app"])
client = make_client(
token=connector_state["write_token"],
audit=audit,
transport=transport,
)
created = await client.create_event("Fereastră de mentenanță")
assert created == {
"id": "event-002",
"title": "Fereastră de mentenanță",
}
@pytest.mark.asyncio
async def test_unknown_token_is_rejected(audit, connector_state):
transport = httpx.ASGITransport(app=connector_state["app"])
client = make_client(
token="unknown-" + os.urandom(24).hex(),
audit=audit,
transport=transport,
)
with pytest.raises(ConnectorRejected) as captured:
await client.list_events()
assert captured.value.status_code == 401
assert captured.value.detail == "token is unknown or revoked"
assert "invalid_token" in (captured.value.challenge or "")
@pytest.mark.asyncio
async def test_revoked_token_stops_working(audit, connector_state):
token = connector_state["write_token"]
assert connector_state["store"].revoke(token) is True
transport = httpx.ASGITransport(app=connector_state["app"])
client = make_client(
token=token,
audit=audit,
transport=transport,
)
with pytest.raises(ConnectorRejected) as captured:
await client.create_event("Nu trebuie creat")
assert captured.value.status_code == 401
assert captured.value.detail == "token is unknown or revoked"
@pytest.mark.asyncio
async def test_transport_failure_is_not_retried(audit):
calls = 0
async def fail_with_timeout(request: httpx.Request) -> httpx.Response:
nonlocal calls
calls += 1
raise httpx.ReadTimeout(
"simulated connector timeout",
request=request,
)
client = make_client(
token="timeout-" + os.urandom(24).hex(),
audit=audit,
transport=httpx.MockTransport(fail_with_timeout),
)
with pytest.raises(
ConnectorUnavailable,
match="action not confirmed",
):
await client.create_event("Operație cu rezultat incert")
assert calls == 1
@pytest.mark.asyncio
async def test_audit_is_valid_and_excludes_raw_token(
audit,
connector_state,
):
token = connector_state["read_token"]
transport = httpx.ASGITransport(app=connector_state["app"])
client = make_client(
token=token,
audit=audit,
transport=transport,
)
await client.list_events()
assert audit.verify() > 0
assert token not in audit.path.read_text(encoding="utf-8")
def test_audit_detects_tampering(audit, tmp_path):
assert audit.verify() > 0
altered_path = tmp_path / "altered-audit.jsonl"
content = audit.path.read_text(encoding="utf-8")
altered_path.write_text(
content.replace(
'"outcome": "success"',
'"outcome": "changed"',
1,
),
encoding="utf-8",
)
altered = AuditTrail(altered_path, audit.key)
with pytest.raises(
ValueError,
match="audit chain broken at sequence",
):
altered.verify()Tokenurile sunt generate cu os.urandom la fiecare rulare și nu sunt scrise în repository. Testul de revocare folosește același token înainte și după schimbarea stării registrului. Cel de timeout numără apelurile și cere exact unul, probând că operația de scriere nu este reluată.
Alterarea jurnalului se face numai pe o copie temporară. Fișierul original rămâne valid și poate fi verificat la închiderea fixture-ului de sesiune.
Punct de control: dacă un test de refuz nu primește 401 sau 403, nu modifica așteptarea pentru a obține verde; verifică mai întâi scope-ul, starea tokenului și antetul Authorization.
Pasul 6 — Păstrăm raportul automat și oprim rularea la prima problemă de mediu
Fișierul scripts/run_suite.sh:
#!/usr/bin/env bash
set -euo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$ROOT_DIR"
if [[ ! -x ".venv/bin/python" ]]; then
printf 'Lipsește mediul virtual: %s/.venvn' "$ROOT_DIR" >&2
exit 1
fi
export PYTHONPATH="$ROOT_DIR/src"
mkdir -p artifacts
rm -f artifacts/audit.jsonl artifacts/junit.xml
.venv/bin/python -m pip check
.venv/bin/python -m pytest
--quiet
--junitxml=artifacts/junit.xml
.venv/bin/python -c
'from pathlib import Path; from connector_lab.audit import AuditTrail; print("Audit generat:", Path("artifacts/audit.jsonl").resolve())'
printf 'Raport JUnit: %s/artifacts/junit.xmln' "$ROOT_DIR"Acordă drept de execuție și pornește suita:
chmod 750 scripts/run_suite.sh
./scripts/run_suite.shset -euo pipefail oprește scriptul la o comandă eșuată, la folosirea unei variabile nedefinite sau la eșecul unei componente dintr-un pipeline. PYTHONPATH indică explicit directorul cu pachetul local.
Exemplu de rezultat; timpul și calea absolută diferă între sisteme:
No broken requirements found.
........ [100%]
- generated xml file: .../artifacts/junit.xml -
8 passed in 0.xx s
Audit generat: .../artifacts/audit.jsonl
Raport JUnit: .../artifacts/junit.xmlPunct de control: rezultatul acceptat este 8 passed, existența ambelor artefacte și codul de ieșire zero al scriptului.
Testul de capăt confirmă permisiunile, revocarea și lanțul HMAC
Rulează o verificare independentă a raportului și a formei logului:
set -euo pipefail
test -s artifacts/junit.xml
test -s artifacts/audit.jsonl
grep -q 'tests="8"' artifacts/junit.xml
grep -q '"outcome": "denied"' artifacts/audit.jsonl
grep -q '"outcome": "transport_error"' artifacts/audit.jsonl
grep -q '"status_code": 403' artifacts/audit.jsonl
grep -q '"status_code": 401' artifacts/audit.jsonl
printf 'Contract verificat: 8 teste, refuzuri și eșec controlat.n'Output-ul așteptat este:
Contract verificat: 8 teste, refuzuri și eșec controlat.În artifacts/audit.jsonl, fiecare rând trebuie să fie JSON valid și să conțină sequence, timestamp, action, actor, token_fingerprint, outcome, status_code, previous_hash și entry_hash. Nu copia jurnalul într-un ticket public: chiar dacă nu include tokenurile brute, poate dezvălui acțiuni, identități interne și momente operaționale.
Testarea conectorului nu înlocuiește testarea prompt injection. Conținutul primit din calendar poate deveni intrare neîncrezătoare pentru model; verificarea trebuie făcută separat, într-un laborator defensiv de prompt injection.
Cinci defecțiuni care localizează ruptura contractului
connector rejected request with status 403: required scope: calendar.write
Cauza este folosirea tokenului de citire la POST /calendar/events. Acesta este rezultatul corect al testului negativ. Dacă apare în testul de scriere permisă, verifică în fixture că write_token este înregistrat cu ambele scope-uri:
grep -n 'calendar.write' tests/test_contract.pyNu rezolva eroarea acordând calendar.write tuturor tokenurilor.
connector rejected request with status 401: token is unknown or revoked
În testul de revocare, mesajul este așteptat. Dacă apare la primul test, tokenul transmis clientului nu este același cu cel înregistrat în TokenStore. Rulează izolat proba:
PYTHONPATH=src python -m pytest
tests/test_contract.py::test_read_scope_lists_events
-vvconnector unavailable; action not confirmed
Mesajul este produs de httpx.TransportError. În testul dedicat confirmă oprirea sigură. Într-o integrare reală, executorul trebuie să marcheze acțiunea drept incertă și să nu repete automat scrierea. Pentru retry sigur este necesară o cheie de idempotency acceptată și memorată de serviciul destinație; laboratorul nu presupune că acel mecanism există.
ValueError: audit chain broken at sequence N
Un rând a fost schimbat, eliminat sau reordonat, ori jurnalul este verificat cu altă cheie. În testul de alterare, mesajul este rezultatul urmărit. Dacă apare pentru artifacts/audit.jsonl, regenerează artefactele prin suita completă:
./scripts/run_suite.shNu edita manual jurnalul pentru a reface hash-urile.
ModuleNotFoundError: No module named 'connector_lab'
Comanda a fost lansată fără directorul src în calea de import sau din alt proiect. Folosește scriptul, care fixează singur directorul rădăcină:
cd "$HOME/connector-contract-lab"
./scripts/run_suite.shEliminăm laboratorul fără să atingem alte medii Python
Comenzile următoare șterg întregul proiect, inclusiv mediul virtual și artefactele. Verifică valoarea afișată înainte de rm -rf.
set -euo pipefail
PROJECT="$HOME/connector-contract-lab"
printf 'Va fi șters numai: %sn' "$PROJECT"
test "$PROJECT" = "$HOME/connector-contract-lab"
test -f "$PROJECT/scripts/run_suite.sh"
deactivate 2>/dev/null || true
rm -rf -- "$PROJECT"Pentru că instalarea s-a făcut într-un mediu virtual dedicat, nu trebuie eliminate pachete din Python-ul sistemului. Laboratorul nu deschide porturi, nu modifică firewall-ul și nu persistă servicii.
Lista de acceptare înaintea unui conector real
- ☐ Tokenul cu
calendar.readpoate executa numai citirea. - ☐ Același token primește
403la operația de scriere. - ☐ Antetul de refuz indică
insufficient_scopeșicalendar.write. - ☐ Tokenul necunoscut primește
401șiinvalid_token. - ☐ Tokenul revocat nu mai funcționează la următorul apel.
- ☐ O eroare de transport produce
action not confirmed. - ☐ Scrierea care eșuează la transport este încercată o singură dată.
- ☐ Jurnalul conține succesul, refuzul și eroarea de transport.
- ☐ Tokenurile brute nu apar în
audit.jsonl. - ☐ Modificarea unui rând rupe verificarea lanțului HMAC.
- ☐ Raportul JUnit confirmă toate cele opt teste.
Următorul proiect logic este înlocuirea registrului local cu un server OAuth controlat de organizație și verificarea revocării prin mecanismul furnizorului. După aceea, păstrează aceeași suită ca test de contract în CI și adaugă un mediu staging cu un cont fără date reale. Pentru conectorii care suportă idempotency, extinde cazul de timeout cu o cheie unică și verifică faptul că două cereri cu aceeași cheie produc o singură acțiune.
Întrebări frecvente
De ce nu folosim un model AI în teste?
Scopul este măsurarea deterministă a graniței dintre executorul agentului și conector. Alegerea tool-ului de către model și rezistența la prompt injection trebuie evaluate separat.
De ce operația de scriere nu este reluată după timeout?
Conectorul ar fi putut procesa cererea înainte să se piardă răspunsul. Fără un mecanism de idempotency, repetarea automată poate crea acțiunea de două ori.
Hash-ul tokenului din registru înseamnă că tokenul poate fi slab?
Nu. Hash-ul limitează expunerea accidentală, dar tokenul trebuie în continuare generat aleatoriu, transmis securizat și administrat de un sistem de autorizare.
Jurnalul HMAC este imuabil?
Nu împotriva unei persoane care deține și cheia, și acces de scriere la fișier. În producție, cheia și destinația centrală de audit trebuie separate de aplicație și protejate prin drepturi distincte.
