GTI_GERENCIADOR_FISCAL/central/tests/test_rtc_security.py

"""Security review of the centrally controlled cClassTrib write: who can change the mode, what the Agent can and cannot decide, tenant scoping, no secrets in audits."""
from sqlalchemy import select
from sqlalchemy.orm import Session

from central.app.models import AuditEvent, Installation, RtcCandidate, RtcWriteAudit
from central.tests.test_control_panel_pages import _panel_client, _seed_rtc
from central.tests.test_zero_touch_hub import H, OPERATIONAL, _client, age_last_sync, finish, hb


def test_new_rtc_routes_require_an_operator_session():
    client, agent_id, _ = _panel_client()
    client.cookies.clear()
    for method, path in (("get", f"/agents/{agent_id}/rtc"), ("post", f"/agents/{agent_id}/rtc-mode"), ("post", f"/agents/{agent_id}/rtc-candidates"), ("post", f"/agents/{agent_id}/rtc-rollback")):
        response = getattr(client, method)(path, follow_redirects=False)
        assert response.status_code in (303, 401, 403), (path, response.status_code)
        assert "login" in response.headers.get("location", "login") or response.status_code in (401, 403)


def test_default_mode_is_disabled_and_the_agent_cannot_choose_it():
    client, engine = _client()
    hb(client)
    with Session(engine) as s:
        assert s.scalars(select(Installation)).one().rtc_write_mode == "DISABLED"
    job = client.get("/v1/jobs/next", headers=H).json()
    assert job["authorizations"]["rtc_write_mode"] == "DISABLED"
    # a forged result cannot change the mode, approve candidates or request rollbacks: none of those are read from the payload
    forged = {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "rtc_write_mode": "ENABLED", "authorizations": {"rtc_write_mode": "ENABLED", "rtc_approved_codes": ["X"]},
              "rtc_auto_apply": {"mode": "ENABLED", "hub_mode": "ENABLED", "preview": [{"product_code": "X", "proposed": "200003", "rule_id": "RTC-R3", "status": "APPROVED"}]}}
    client.post(f"/v1/jobs/{job['job_id']}/complete", json={"result_summary": forged}, headers=H)
    with Session(engine) as s:
        assert s.scalars(select(Installation)).one().rtc_write_mode == "DISABLED"
        assert s.scalars(select(RtcCandidate)).all() == []                                    # DISABLED stores no preview at all


def test_preview_from_the_agent_never_arrives_approved_and_never_approves_itself():
    client, engine = _client()
    hb(client)
    with Session(engine) as s:
        s.scalars(select(Installation)).one().rtc_write_mode = "APPROVAL_REQUIRED"
        s.commit()
    item = {"product_code": "X1", "description": "D", "ncm": "10063021", "current": "", "proposed": "200003", "cst": "200", "rule_id": "RTC-R3", "basis": "b", "status": "APPROVED"}
    finish(client, engine, {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "fiscal_context": {"evaluated_all": True}, "rtc_auto_apply": {"preview": [item]}})
    with Session(engine) as s:
        assert [c.status for c in s.scalars(select(RtcCandidate)).all()] == ["PENDING"]
    age_last_sync(engine, 4000); hb(client)
    assert client.get("/v1/jobs/next", headers=H).json()["authorizations"]["rtc_approved_codes"] == []


def test_a_rollback_result_cannot_touch_another_installations_audit():
    client, engine = _client()
    hb(client)
    with Session(engine) as s:
        inst = s.scalars(select(Installation)).one()
        foreign = RtcWriteAudit(tenant_id=inst.tenant_id, installation_id=inst.id + 999, agent_id=inst.agent_id, product_code="F", field="CCLASSTRIB", outcome="APPLIED", new_value="200003", action="APPLY", rollback_state="REQUESTED")
        # the FK is not enforced by the SQLite test DB; the scoping is what is under test
        s.add(foreign); s.commit(); foreign_id = foreign.id
    done = {"outcome": "ROLLED_BACK", "product_code": "F", "field": "CCLASSTRIB", "old_value": "200003", "new_value": None, "action": "ROLLBACK", "rollback_of": foreign_id}
    finish(client, engine, {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "rtc_auto_apply": {"rollbacks": [done]}})
    with Session(engine) as s:
        assert s.get(RtcWriteAudit, foreign_id).rollback_state == "REQUESTED"                 # untouched


def test_rollback_and_candidate_actions_are_scoped_to_the_operators_installation_and_audited():
    client, agent_id, _ = _panel_client()
    _seed_rtc(client, agent_id, n=3)
    csrf = client.cookies.get("gti_csrf")
    with client._session_factory() as s:
        other = RtcWriteAudit(tenant_id=1, installation_id=987654, agent_id=agent_id, product_code="Z", field="CCLASSTRIB", outcome="APPLIED", new_value="200003", action="APPLY")
        s.add(other); s.commit(); other_id = other.id
    assert client.post(f"/agents/{agent_id}/rtc-rollback", data={"csrf_token": csrf, "audit_id": other_id, "confirm": "REVERTER"}).status_code == 409
    client.post(f"/agents/{agent_id}/rtc-candidates?all=1", data={"csrf_token": csrf, "action": "approve"}, follow_redirects=False)
    with client._session_factory() as s:
        actions = [a.action for a in s.scalars(select(AuditEvent)).all()]
        assert "RTC_CANDIDATES_APPROVE" in actions


def test_audit_rows_never_carry_credentials_or_tokens():
    client, engine = _client()
    hb(client)
    applied = {"outcome": "APPLIED", "product_code": "10", "field": "CCLASSTRIB", "old_value": None, "new_value": "200003", "rule_id": "RTC-R3", "action": "APPLY", "password": "p", "token": "t",
               "evidence": {"reason": "r"}}
    finish(client, engine, {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "rtc_auto_apply": {"applied": [applied]}})
    with Session(engine) as s:
        row = s.scalars(select(RtcWriteAudit)).one()
        blob = repr([getattr(row, c.name) for c in RtcWriteAudit.__table__.columns])
        assert "password" not in blob.lower() and "'t'" not in blob and "Bearer" not in blob
        assert not any("pass" in c.name or "token" in c.name or "secret" in c.name for c in RtcWriteAudit.__table__.columns)


def test_agent_can_reread_only_its_own_current_authorization_and_it_follows_the_operator():
    client, engine = _client()
    hb(client)
    assert client.get("/v1/agents/me/authorizations").status_code in (401, 403)                     # requires the Agent credential
    assert client.get("/v1/agents/me/authorizations", headers=H).json()["rtc_write_mode"] == "DISABLED"
    with Session(engine) as s:
        s.scalars(select(Installation)).one().rtc_write_mode = "ENABLED"
        s.commit()
    body = client.get("/v1/agents/me/authorizations", headers=H).json()
    assert body["rtc_write_mode"] == "ENABLED" and set(body) == {"rtc_write_mode", "rtc_approved_codes", "rtc_rollbacks", "features", "cfop_ippt_write_mode"}


def _become(client, role):
    """Turn the logged-in test operator into a Gerenciador SSO collaborator with the given perfil (identity source of truth is the SSO)."""
    from central.app.models import AdminUser
    with client._session_factory() as s:
        admin = s.scalars(select(AdminUser)).first()
        admin.auth_source, admin.password_hash, admin.external_subject, admin.external_role = "GTI_GERENCIADOR", None, "sso-subject-1", role
        s.commit()


def _post(client, path, **data):
    return client.post(path, data={"csrf_token": client.cookies.get("gti_csrf"), **data}, follow_redirects=False)


def test_write_modes_are_restricted_to_gerenciador_admins_and_every_denial_is_audited():
    client, agent_id, _ = _panel_client()
    _become(client, "PADRAO")
    for mode, extra in (("ENABLED", {"confirm": "PREENCHER"}), ("APPROVAL_REQUIRED", {})):
        assert _post(client, f"/agents/{agent_id}/rtc-mode", mode=mode, **extra).status_code == 403
    assert _post(client, f"/agents/{agent_id}/rtc-candidates", action="approve", all="1").status_code == 403
    assert _post(client, f"/agents/{agent_id}/rtc-candidates", action="apply_approved").status_code == 403
    assert _post(client, f"/agents/{agent_id}/rtc-rollback", audit_id="1", confirm="REVERTER").status_code == 403
    with client._session_factory() as s:
        assert s.scalars(select(Installation)).one().rtc_write_mode == "DISABLED"
        assert len([e for e in s.scalars(select(AuditEvent)) if e.action == "RTC_DENIED"]) == 5
    # read-only modes stay available to a normal collaborator, and the page tells them why the rest is disabled
    assert _post(client, f"/agents/{agent_id}/rtc-mode", mode="DRY_RUN").status_code == 303
    page = client.get(f"/agents/{agent_id}/rtc").text
    assert "Somente leitura para o seu perfil" in page and "disabled" in page


def test_a_normal_collaborator_cannot_switch_off_a_privileged_mode_either():
    client, agent_id, _ = _panel_client()
    with client._session_factory() as s:
        s.scalars(select(Installation)).one().rtc_write_mode = "ENABLED"
        s.commit()
    _become(client, "PADRAO")
    assert _post(client, f"/agents/{agent_id}/rtc-mode", mode="DISABLED").status_code == 403
    with client._session_factory() as s:
        assert s.scalars(select(Installation)).one().rtc_write_mode == "ENABLED"


def test_gerenciador_admin_can_change_modes_and_the_operator_is_recorded():
    client, agent_id, _ = _panel_client()
    _become(client, "ADMIN")
    assert _post(client, f"/agents/{agent_id}/rtc-mode", mode="ENABLED", confirm="PREENCHER").status_code == 303
    with client._session_factory() as s:
        assert s.scalars(select(Installation)).one().rtc_write_mode == "ENABLED"
        event = [e for e in s.scalars(select(AuditEvent)) if e.action == "RTC_WRITE_MODE"][-1]
        assert event.context["to"] == "ENABLED" and event.context["role"] == "ADMIN" and event.context["operator"]