gtiandClaude Sonnet 5
9f9e06e9e9
Replaces the native list with a combobox (name/code/CNPJ search, accent- and case-insensitive, keyboard + mouse, custom dropdown, hidden client_id = the Gerenciador's real CLIENTE.ID). The backend resolves client_id against the active GTI_GERENCIADOR mirror and rejects free text/unknown ids. Adds Tenant.external_cnpj (migration 0025) mirrored via the internal endpoint. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
428 lines
19 KiB
Python
428 lines
19 KiB
Python
"""Install-code enrollment flow: POST /v1/agents/enroll with install_code.
|
|
|
|
Covers the universal-EXE enrollment path — the operator types a 16-char
|
|
XXXX-XXXX-XXXX-XXXX installation code into Setup.exe, the Hub resolves the
|
|
installation by the hash on its own (no installation_id, no enrollment.json),
|
|
issues the permanent agent credential, and exhausts the code (replay-proof).
|
|
Also covers rejection when code and legacy token are sent together, expiry,
|
|
attempts/rate-limit and no-plaintext persistencen.
|
|
"""
|
|
import dataclasses
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import zipfile
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from central.app import config, enrollment, rate_limit, release_storage
|
|
from central.app.main import create_app
|
|
from central.app.models import Agent, AgentEnrollment, AgentRelease, AuditEvent, Installation, Tenant
|
|
from central.scripts.publish_agent_release import publish
|
|
from central.tests.test_control_plane import seed, setup_db
|
|
|
|
VERSION = "1.1.1"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_rate_limit():
|
|
rate_limit._attempts.clear()
|
|
yield
|
|
rate_limit._attempts.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def hub(tmp_path, monkeypatch):
|
|
replaced = dataclasses.replace(
|
|
config.settings,
|
|
agent_release_storage_path=str(tmp_path / "releases"),
|
|
agent_enrollment_ttl_seconds=86400,
|
|
base_url="https://gtifiscal.example.gti",
|
|
web_cookie_secure=False,
|
|
)
|
|
monkeypatch.setattr(config, "settings", replaced)
|
|
if "app.config" in sys.modules:
|
|
monkeypatch.setattr(sys.modules["app.config"], "settings", replaced)
|
|
factory = setup_db()
|
|
seed(factory)
|
|
client = TestClient(create_app(database_probe=lambda: True, session_factory=factory, bootstrap_token=None))
|
|
return client, factory
|
|
|
|
|
|
def login(client):
|
|
response = client.post("/login", data={"username": "operator", "password": "a-secure-password"}, follow_redirects=False)
|
|
assert response.status_code == 303
|
|
|
|
|
|
def csrf(client):
|
|
return client.cookies.get("gti_csrf", "")
|
|
|
|
|
|
@pytest.fixture
|
|
def release_id(hub, tmp_path):
|
|
_client, factory = hub
|
|
data = io_bytes_tmp(tmp_path, VERSION)
|
|
with factory() as session:
|
|
release, _created = publish(session, data, version=VERSION)
|
|
return release.id
|
|
|
|
|
|
def io_bytes_tmp(tmp_path, version):
|
|
from pathlib import Path
|
|
buf = io_bytes_bundle(version)
|
|
path = Path(tmp_path) / f"GTIFiscalAgent-{version}.zip"
|
|
path.write_bytes(buf)
|
|
return path
|
|
|
|
|
|
def io_bytes_bundle(version):
|
|
import io
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as z:
|
|
z.writestr(f"GTIFiscalAgent-{version}/gti-agent.ps1", "# fake\n")
|
|
z.writestr(f"GTIFiscalAgent-{version}/VERSION", version)
|
|
return buf.getvalue()
|
|
|
|
|
|
def tenant_id(factory):
|
|
with factory() as session:
|
|
tenant = session.scalar(select(Tenant).where(Tenant.name == "Acme"))
|
|
# /agents/new only offers active GTI_GERENCIADOR-sourced tenants
|
|
# (central/app/client_sync.py) — flip the shared fixture tenant so
|
|
# existing form-based tests keep working under that rule.
|
|
if tenant.source != "GTI_GERENCIADOR" or not tenant.active:
|
|
tenant.source, tenant.external_client_id, tenant.active = "GTI_GERENCIADOR", tenant.id, True
|
|
session.commit()
|
|
return tenant.id
|
|
|
|
|
|
def make_installation(factory, release_id, name="Servidor E2E"):
|
|
with factory() as session:
|
|
enr = AgentEnrollment(enrollment_uuid=str(__import__("uuid").uuid4()), tenant_id=tenant_id(factory),
|
|
display_name=name, release_id=release_id, status="PENDING")
|
|
session.add(enr)
|
|
session.commit()
|
|
return enr.enrollment_uuid
|
|
|
|
|
|
def set_code(factory, uuid, code, ttl_seconds=86400):
|
|
with factory() as session:
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == uuid))
|
|
enr.install_code_hash = enrollment.install_code_digest(code)
|
|
enr.install_code_expires_at = datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)
|
|
enr.install_code_created_at = datetime.now(timezone.utc)
|
|
session.commit()
|
|
|
|
|
|
def audits(factory):
|
|
with factory() as session:
|
|
return session.scalars(select(AuditEvent)).all()
|
|
|
|
|
|
# ---------- happy path: code alone resolves the installation ----------
|
|
|
|
def test_enroll_with_install_code_alone_creates_agent(hub, release_id):
|
|
client, factory = hub
|
|
uuid = make_installation(factory, release_id)
|
|
code = enrollment.generate_install_code()
|
|
set_code(factory, uuid, code)
|
|
|
|
enrolled = client.post("/v1/agents/enroll", json={"install_code": code, "agent_version": "1.1.1", "hostname": "WIN-E2E", "platform": "windows"})
|
|
assert enrolled.status_code == 200
|
|
body = enrolled.json()
|
|
assert body["agent_id"] and body["agent_token"] and body["hostname_mismatch"] is False
|
|
assert body["installation_id"] == uuid
|
|
|
|
with factory() as session:
|
|
agent = session.get(Agent, body["agent_id"])
|
|
assert agent.enabled and agent.agent_version == "1.1.1"
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
assert installation.installation_id == uuid and installation.hostname == "WIN-E2E"
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == uuid))
|
|
assert enr.status == "ENROLLED" and enr.agent_id == agent.id
|
|
assert enr.install_code_hash is None and enr.install_code_expires_at is None # code consumed, replay-proof
|
|
|
|
|
|
def test_replay_of_consumed_code_is_rejected(hub, release_id):
|
|
client, factory = hub
|
|
uuid = make_installation(factory, release_id)
|
|
code = enrollment.generate_install_code()
|
|
set_code(factory, uuid, code)
|
|
first = client.post("/v1/agents/enroll", json={"install_code": code})
|
|
assert first.status_code == 200
|
|
second = client.post("/v1/agents/enroll", json={"install_code": code})
|
|
assert second.status_code == 401
|
|
|
|
events = {e.action for e in audits(factory)}
|
|
assert "ENROLLMENT_SUCCEEDED" in events and "ENROLLMENT_REJECTED" in events
|
|
raw = json.dumps({"action": "ENROLLMENT_SUCCEEDED", "target": ["x"], "context": {"ip": "y"}}) + code
|
|
# the code plaintext is never in any audit event context
|
|
for event in audits(factory):
|
|
assert code not in json.dumps(event.context)
|
|
assert enrollment.normalize_install_code(code) not in json.dumps(event.context)
|
|
|
|
|
|
def test_code_is_case_and_hyphen_insensitive(hub, release_id):
|
|
client, factory = hub
|
|
uuid = make_installation(factory, release_id)
|
|
code = enrollment.generate_install_code()
|
|
set_code(factory, uuid, code)
|
|
typed = code.lower().replace("-", " ")
|
|
enrolled = client.post("/v1/agents/enroll", json={"install_code": typed})
|
|
assert enrolled.status_code == 200
|
|
|
|
|
|
# ---------- rejections ----------
|
|
|
|
def test_invalid_code_format_rejected_without_lookup(hub, release_id):
|
|
client, factory = hub
|
|
invalid = client.post("/v1/agents/enroll", json={"install_code": "NOT-A-CODE"})
|
|
assert invalid.status_code == 400 or invalid.status_code == 401
|
|
assert "INVALID" in str(invalid.json())
|
|
|
|
|
|
def test_unknown_but_wellformed_code_rejected(hub, release_id):
|
|
client, factory = hub
|
|
# valid-format code that was never issued for this Hub
|
|
made_up = "-".join(["0123"] * 4)
|
|
rejected = client.post("/v1/agents/enroll", json={"install_code": made_up})
|
|
assert rejected.status_code == 401
|
|
assert "INVALID" in str(rejected.json())
|
|
|
|
|
|
def test_expired_code_rejected(hub, release_id):
|
|
client, factory = hub
|
|
uuid = make_installation(factory, release_id)
|
|
code = enrollment.generate_install_code()
|
|
set_code(factory, uuid, code, ttl_seconds=-60)
|
|
rejected = client.post("/v1/agents/enroll", json={"install_code": code})
|
|
assert rejected.status_code == 401
|
|
assert "EXPIRED" in str(rejected.json())
|
|
|
|
|
|
def test_code_for_revoked_installation_rejected(hub, release_id):
|
|
client, factory = hub
|
|
uuid = make_installation(factory, release_id)
|
|
code = enrollment.generate_install_code()
|
|
set_code(factory, uuid, code)
|
|
with factory() as session:
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == uuid))
|
|
enr.status = "REVOKED"
|
|
session.commit()
|
|
rejected = client.post("/v1/agents/enroll", json={"install_code": code})
|
|
assert rejected.status_code == 401
|
|
assert "REVOKED" in str(rejected.json())
|
|
|
|
|
|
def test_code_and_token_together_rejected(hub, release_id):
|
|
client, factory = hub
|
|
uuid = make_installation(factory, release_id)
|
|
code = enrollment.generate_install_code()
|
|
set_code(factory, uuid, code)
|
|
rejected = client.post("/v1/agents/enroll", json={
|
|
"install_code": code, "installation_id": uuid, "enrollment_token": "x" * 43,
|
|
})
|
|
assert rejected.status_code == 400
|
|
|
|
|
|
def test_failed_attempts_increment_and_rate_limit(hub, release_id):
|
|
client, factory = hub
|
|
uuid = make_installation(factory, release_id)
|
|
code = enrollment.generate_install_code()
|
|
set_code(factory, uuid, code)
|
|
# several wrong-format tries that never reach the code lookup
|
|
for _ in range(12):
|
|
client.post("/v1/agents/enroll", json={"install_code": "NOT-A-CODE"})
|
|
assert rate_limit.blocked("enroll", "testclient")
|
|
|
|
with factory() as session:
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == uuid))
|
|
assert enr.install_code_attempts >= 0 # format failures don't consume this installation's code attempts
|
|
|
|
|
|
def test_wrong_code_rejected_as_invalid(hub, release_id):
|
|
client, factory = hub
|
|
uuid = make_installation(factory, release_id)
|
|
right_code = enrollment.generate_install_code()
|
|
set_code(factory, uuid, right_code)
|
|
wrong = enrollment.generate_install_code() # well-formed but never issued
|
|
attempt = client.post("/v1/agents/enroll", json={"install_code": wrong})
|
|
assert attempt.status_code == 401
|
|
assert "INVALID" in str(attempt.json())
|
|
|
|
|
|
def test_resolved_code_failures_increment_attempt_budget(hub, release_id):
|
|
# A well-formed code that resolves to a REVOKED installation consumes the
|
|
# installation's attempt budget (per-installation rate limit), same as an
|
|
# expired one.
|
|
client, factory = hub
|
|
uuid = make_installation(factory, release_id)
|
|
code = enrollment.generate_install_code()
|
|
set_code(factory, uuid, code)
|
|
with factory() as session:
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == uuid))
|
|
enr.status = "REVOKED"
|
|
session.commit()
|
|
for _ in range(3):
|
|
client.post("/v1/agents/enroll", json={"install_code": code})
|
|
with factory() as session:
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == uuid))
|
|
assert enr.install_code_attempts >= 3
|
|
|
|
|
|
def test_legacy_token_path_still_works(hub, release_id):
|
|
client, factory = hub
|
|
uuid = make_installation(factory, release_id)
|
|
token = enrollment.new_token()
|
|
with factory() as session:
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == uuid))
|
|
enr.enrollment_token_hash = enrollment.token_digest(token)
|
|
enr.enrollment_token_expires_at = datetime.now(timezone.utc) + timedelta(seconds=86400)
|
|
session.commit()
|
|
enrolled = client.post("/v1/agents/enroll", json={"installation_id": uuid, "enrollment_token": token})
|
|
assert enrolled.status_code == 200
|
|
|
|
|
|
# ---------- publish EXE releases (immutable) ----------
|
|
|
|
def test_publish_exe_release(hub, tmp_path):
|
|
client, factory = hub
|
|
from central.app import release_storage
|
|
exe_bytes = b"MZ\x90\x00 fake windows installer exe body \x00" * 100
|
|
exe_path = Path(tmp_path) / "GTIFiscalAgent-Setup-1.1.1.exe"
|
|
exe_path.write_bytes(exe_bytes)
|
|
with factory() as session:
|
|
release, created = publish(session, exe_path, version="1.1.1", notes="universal installer")
|
|
assert created and release.status == "PUBLISHED" and release.published_at is not None
|
|
assert release.filename == "GTIFiscalAgent-Setup-1.1.1.exe"
|
|
assert release_storage.safe_path_for(release.storage_path).read_bytes() == exe_bytes
|
|
|
|
|
|
def test_publish_exe_is_immutable(hub, tmp_path):
|
|
_client, factory = hub
|
|
exe_path = Path(tmp_path) / "GTIFiscalAgent-Setup-1.1.1.exe"
|
|
exe_path.write_bytes(b"MZ fake body v1")
|
|
with factory() as session:
|
|
publish(session, exe_path, version="1.1.1")
|
|
exe_path.write_bytes(b"MZ DIFFERENT body v2")
|
|
with factory() as session:
|
|
with pytest.raises(SystemExit):
|
|
publish(session, exe_path, version="1.1.1") # refused: immutable once PUBLISHED
|
|
|
|
|
|
def test_publish_identical_exe_is_noop(hub, tmp_path):
|
|
_client, factory = hub
|
|
exe_path = Path(tmp_path) / "GTIFiscalAgent-Setup-1.1.1.exe"
|
|
exe_path.write_bytes(b"MZ same bytes")
|
|
with factory() as session:
|
|
release, created = publish(session, exe_path, version="1.1.1")
|
|
assert created
|
|
with factory() as session:
|
|
release2, created2 = publish(session, exe_path, version="1.1.1")
|
|
assert release2.id == release.id and not created2
|
|
|
|
|
|
# ---------- EXE download flow (panel: code -> download -> enroll) ----------
|
|
|
|
def publish_exe_release(factory, tmp_path):
|
|
exe_path = tmp_path / "GTIFiscalAgent-Setup-1.1.1.exe"
|
|
exe_path.write_bytes(b"MZ\x90\x00 GTIFiscalAgent installer 1.1.1 \x00" * 40)
|
|
with factory() as session:
|
|
release, _created = publish(session, exe_path, version="1.1.1")
|
|
return release.id
|
|
|
|
|
|
def test_panel_flow_code_generate_download_and_enroll_with_exe(hub, tmp_path):
|
|
client, factory = hub
|
|
release_id = publish_exe_release(factory, tmp_path)
|
|
login(client)
|
|
created = client.post("/agents/new", data={"csrf_token": csrf(client), "client_id": tenant_id(factory),
|
|
"display_name": "Servidor EXE", "release_id": release_id}, follow_redirects=False)
|
|
assert created.status_code == 303
|
|
installation_uuid = created.headers["location"].rsplit("/", 1)[-1]
|
|
|
|
# generate the install code (only hash stored; plaintext shown once via ?code=)
|
|
code_page = client.post(f"/agents/pending/{installation_uuid}/code", data={"csrf_token": csrf(client)}, follow_redirects=True)
|
|
assert code_page.status_code == 200
|
|
# the redirect carried ?code=XXXX-... ; the page must show it
|
|
code_addr = code_page.history[0].headers.get("location") if code_page.history else ""
|
|
assert "code=" in code_addr
|
|
install_code = code_addr.rsplit("code=", 1)[-1].split("&")[0]
|
|
assert len(install_code) == 19
|
|
|
|
with factory() as session:
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == installation_uuid))
|
|
assert enr.install_code_hash == enrollment.install_code_digest(install_code)
|
|
assert enr.install_code_expires_at is not None
|
|
assert enr.install_code_hash != install_code # never plaintext in DB
|
|
|
|
# download the EXE from the pending page (authenticated direct download)
|
|
downloaded = client.get(f"/agents/pending/{installation_uuid}/download")
|
|
assert downloaded.status_code == 200
|
|
assert downloaded.content.startswith(b"MZ")
|
|
assert "attachment" in downloaded.headers["content-disposition"]
|
|
assert "GTIFiscalAgent-Setup-1.1.1.exe" in downloaded.headers["content-disposition"]
|
|
|
|
# enroll with just the code (no installation id, no token)
|
|
anon = TestClient(client.app)
|
|
enrolled = anon.post("/v1/agents/enroll", json={"install_code": install_code, "agent_version": "1.1.1", "hostname": "WIN-EXE"})
|
|
assert enrolled.status_code == 200
|
|
body = enrolled.json()
|
|
assert body["agent_id"] and body["agent_token"]
|
|
|
|
with factory() as session:
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == installation_uuid))
|
|
assert enr.status == "ENROLLED" and enr.install_code_hash is None # consumed
|
|
agent = session.get(Agent, body["agent_id"])
|
|
assert agent.enabled and agent.agent_version == "1.1.1"
|
|
|
|
events = {e.action for e in audits(factory)}
|
|
assert {"INSTALLATION_CREATED", "INSTALLATION_CODE_CREATED", "INSTALLER_DOWNLOADED", "ENROLLMENT_STARTED",
|
|
"ENROLLMENT_SUCCEEDED", "AGENT_CREDENTIAL_ISSUED"} <= events
|
|
for event in audits(factory):
|
|
assert install_code not in json.dumps(event.context)
|
|
|
|
|
|
def test_rotation_generates_a_code_that_invalidates_the_previous(hub, tmp_path):
|
|
client, factory = hub
|
|
release_id = publish_exe_release(factory, tmp_path)
|
|
login(client)
|
|
created = client.post("/agents/new", data={"csrf_token": csrf(client), "client_id": tenant_id(factory),
|
|
"display_name": "Rotação", "release_id": release_id}, follow_redirects=False)
|
|
installation_uuid = created.headers["location"].rsplit("/", 1)[-1]
|
|
first = client.post(f"/agents/pending/{installation_uuid}/code", data={"csrf_token": csrf(client)}, follow_redirects=True)
|
|
first_code = first.history[0].headers["location"].rsplit("code=", 1)[-1].split("&")[0]
|
|
second = client.post(f"/agents/pending/{installation_uuid}/code", data={"csrf_token": csrf(client)}, follow_redirects=True)
|
|
second_code = second.history[0].headers["location"].rsplit("code=", 1)[-1].split("&")[0]
|
|
assert first_code != second_code
|
|
# first code no longer resolves (rotated), second does
|
|
with factory() as session:
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == installation_uuid))
|
|
assert enr.install_code_hash == enrollment.install_code_digest(second_code)
|
|
anon = TestClient(client.app)
|
|
assert anon.post("/v1/agents/enroll", json={"install_code": first_code}).status_code == 401
|
|
assert anon.post("/v1/agents/enroll", json={"install_code": second_code}).status_code == 200
|
|
events = [e.action for e in audits(factory)]
|
|
assert events.count("INSTALLATION_CODE_CREATED") >= 2
|
|
|
|
|
|
def test_revoke_clears_code_and_is_audited(hub, tmp_path):
|
|
client, factory = hub
|
|
release_id = publish_exe_release(factory, tmp_path)
|
|
login(client)
|
|
created = client.post("/agents/new", data={"csrf_token": csrf(client), "client_id": tenant_id(factory),
|
|
"display_name": "Revoga", "release_id": release_id}, follow_redirects=False)
|
|
installation_uuid = created.headers["location"].rsplit("/", 1)[-1]
|
|
client.post(f"/agents/pending/{installation_uuid}/code", data={"csrf_token": csrf(client)})
|
|
revoked = client.post(f"/agents/pending/{installation_uuid}/revoke", data={"csrf_token": csrf(client)}, follow_redirects=False)
|
|
assert revoked.status_code == 303
|
|
with factory() as session:
|
|
enr = session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == installation_uuid))
|
|
assert enr.status == "REVOKED" and enr.install_code_hash is None
|
|
events = {e.action for e in audits(factory)}
|
|
assert "INSTALLATION_CODE_CREATED" in events and "INSTALLATION_CODE_REVOKED" in events and "INSTALLATION_REVOKED" in events
|