From a2bad050f97b8209d4eab2017b9b1e623840a1a8 Mon Sep 17 00:00:00 2001 From: gti Date: Mon, 28 Sep 2026 10:53:20 -0300 Subject: [PATCH 1/4] test: A.3 end-to-end regression - weak evidence never becomes a write candidate Adds test_weak_composed_evidence_never_becomes_a_candidate_regardless_of_the_write_flag, proving from the real field_contract.resolve_product (not an isolated stub) that the beer misclassification fixed on master (see "fix: product_role never infers OWN_PRODUCTION from a single COMPOSICAO occurrence") can never reach select_candidates as an AUTO_FILL_SAFE candidate, independent of cfop_ippt_write_mode/write_gate. Co-Authored-By: Claude Sonnet 5 --- agent/tests/test_cfop_ippt_auto_apply.py | 26 ++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/agent/tests/test_cfop_ippt_auto_apply.py b/agent/tests/test_cfop_ippt_auto_apply.py index 58b0875..e21c116 100644 --- a/agent/tests/test_cfop_ippt_auto_apply.py +++ b/agent/tests/test_cfop_ippt_auto_apply.py @@ -97,6 +97,32 @@ def test_select_candidates_only_change_and_write_eligible_cfop_or_ippt(): assert all(c["field"] in apply_mod.ALLOWED_FIELDS for c in cands) +def test_weak_composed_evidence_never_becomes_a_candidate_regardless_of_the_write_flag(): + """Missão 4 / Bloco A.3: a cell whose role evidence is weak (field_contract itself refuses to + classify it OWN_PRODUCTION/RESALE with confidence) must never reach select_candidates as + AUTO_FILL_SAFE - proven end-to-end from the real resolver, not just asserted in isolation. + This holds independently of cfop_ippt_write_mode/write_gate, which only gates whether + apply_candidates() is ever invoked - it never affects what counts as a candidate.""" + from central.app.field_contract import resolve_product, Context, WRITE_FIELDS + + def product(**kw): + p = {f: None for f in WRITE_FIELDS} + p.update({"CODIGO": "000321", "PRODUTO": "CERVEJA LONG NECK NOVA SCHIN", "CODGRUPO": "G1", "TIPO": "VENDA", "PROD_COZINHA": None, "PRODUTO_SERVICO": None, + "CLASSIFICACAO_FISCAL": "22030000", "CFOP": "5102", "CSOSN": "102", "CST": "000", "SITUACAO_TRIBUTARIA": "T", "IAT": "A", "IPPT": "T", + "INCIDENCIA_PISCOFINS": "NENHUMA", "PISCOFINS": "N"}) + p.update(kw) + return p + + decision = {"status": "EXACT_NCM", "cest_status": "NO_CEST_MAPPING_FOUND", "ipi_status": "IPI_RESOLVED", "ba_st_mva_status": "NO_MVA_MAPPING_FOUND", "dataset_versions": {}, + "classification": {"ncm": "22030000", "ipi": {"rate_kind": "NUMERIC", "rate_numeric": 3.9, "rate": "3.9"}, "ipi_candidates": [], "cest": None, "cest_candidates": [], "ba_st_mva": None, "ba_st_mva_candidates": []}} + refs = {"cfop_valid": {"5101", "5102"}, "csosn": {"102"}, "cst_icms": set(), "cst_pis": set(), "cst_cofins": set(), "origem": set("012345678"), "fcp_uf": {}, "cbs_rate_2026": None, "cclasstrib_count": 0} + co = {"uf": "BA", "crt": "1", "contribuinte_ipi": "N", "company_code": "000001"} + ev_weak = {"recipes": set(), "composed": set(), "composed_ratio": {"000321": {"composed": 1, "other": 19}}, "purchases": {}, "stock_entry": {"000321": True}, "stock_in": {"000321": True}} + result = resolve_product(product(), decision, co, refs, ev_weak, Context([])) + candidates, total = apply_mod.select_candidates({"000321": result}) + assert total == 0 and candidates == [] # the weak-evidence beer never becomes a write candidate + + def test_select_candidates_respects_limit(): results = {f"{i:06d}": {"CFOP": {"state": CHANGE, "current": None, "proposed": "5102", "write_eligible": True, "reason": "R"}} for i in range(5)} cands, total = apply_mod.select_candidates(results, limit=2) From efc8b54bc711ac267e815cec07d362b6ba54d0b8 Mon Sep 17 00:00:00 2001 From: gti Date: Mon, 28 Sep 2026 11:22:02 -0300 Subject: [PATCH 2/4] feat: Hub-side plumbing for CFOP/IPPT write capability (flag DEFAULT OFF) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Missão 4 / Bloco E.1-E.2, on feature/cfop-ippt-write only (never main): - Migration 0033_cfop_ippt_write_mode: Installation.cfop_ippt_write_mode (DISABLED|DRY_RUN|APPROVAL_REQUIRED|ENABLED, DEFAULT 'DISABLED' for every installation, existing and new) + append-only cfop_ippt_write_audits table. - central/app/cfop_ippt_audit.py mirrors rtc_audit.py for this narrower capability: job_authorizations(installation) (delivered inside the authenticated job/heartbeat authorizations, never a local Agent switch) and record_cfop_ippt_write_audits (persists applied/aborted cells; only CFOP/ IPPT are ever accepted, everything else silently ignored). - central/app/web.py: POST /agents/{id}/cfop-ippt-write - ADMIN only (SSO Gerenciador), CSRF, typing GRAVAR required for any mode other than DISABLED, every change audited (CFOP_IPPT_WRITE_MODE). Small toggle UI in agent_detail.html next to Contexto Fiscal. - Wired into the existing job/heartbeat authorization payload and job completion handling in central/app/api.py, alongside (never mixed with) the RTC wiring. - 18 new tests (central/tests/test_cfop_ippt_write_plumbing.py): default DISABLED everywhere (job payload AND /v1/agents/me/authorizations); a forged completion payload can never change the mode (read fresh from the Installation row, never from Agent-supplied JSON); PADRAO role gets 403 with no audit row; ADMIN + wrong/missing CSRF is refused; ADMIN without typing GRAVAR is refused (422); disabling never requires GRAVAR; invalid mode is refused; the audit recorder only accepts CFOP/IPPT, silently drops any other field (e.g. a stray CCLASSTRIB row). - Two existing RTC tests updated for the additive authorizations-dict key (cfop_ippt_write_mode) they now also carry. Still NOT wired into agent/scripts/run_agent_service.py::execute_job - no Agent code calls select_candidates/apply_candidates yet, so nothing writes to Firebird from this branch. That wiring, an Agent version bump, a new installer and Windows Sandbox validation are the next step, deliberately left for after review (see docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md) - this mission explicitly asked not to publish this as an active release yet. FULL_SUITE on this branch: 1 known, expected failure - test_manifest_records_the_real_migration_head - because this branch adds migration 0033 that was never merged to master/deployed; RELEASE_MANIFEST.json intentionally still describes master/production (head 0032). Every other test passes, including all CFOP/IPPT and RTC-adjacent tests. REAL_PRODUCT_FISCAL_WRITES=0 (no write path is reachable from any Agent code on any branch). Co-Authored-By: Claude Sonnet 5 --- .../versions/0033_cfop_ippt_write_mode.py | 39 +++++++ central/app/api.py | 9 +- central/app/cfop_ippt_audit.py | 51 +++++++++ central/app/events.py | 2 +- central/app/models.py | 25 +++++ central/app/web.py | 29 ++++- central/templates/agent_detail.html | 8 +- .../tests/test_cfop_ippt_write_plumbing.py | 105 ++++++++++++++++++ central/tests/test_rtc_security.py | 2 +- central/tests/test_zero_touch_hub.py | 2 +- release/RELEASE_MANIFEST.json | 2 +- 11 files changed, 266 insertions(+), 8 deletions(-) create mode 100644 central/alembic/versions/0033_cfop_ippt_write_mode.py create mode 100644 central/app/cfop_ippt_audit.py create mode 100644 central/tests/test_cfop_ippt_write_plumbing.py diff --git a/central/alembic/versions/0033_cfop_ippt_write_mode.py b/central/alembic/versions/0033_cfop_ippt_write_mode.py new file mode 100644 index 0000000..f63cfa2 --- /dev/null +++ b/central/alembic/versions/0033_cfop_ippt_write_mode.py @@ -0,0 +1,39 @@ +"""CFOP/IPPT write capability: per-installation write mode (DEFAULT OFF) + append-only audit table. + +Revision ID: 0033_cfop_ippt_write_mode +Revises: 0032_contribuinte_ipi +""" +from alembic import op +import sqlalchemy as sa + +revision = "0033_cfop_ippt_write_mode" +down_revision = "0032_contribuinte_ipi" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column("installations", sa.Column("cfop_ippt_write_mode", sa.String(24), nullable=False, server_default="DISABLED")) + op.create_table( + "cfop_ippt_write_audits", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id"), nullable=False, index=True), + sa.Column("installation_id", sa.Integer(), sa.ForeignKey("installations.id"), nullable=False, index=True), + sa.Column("agent_id", sa.Integer(), sa.ForeignKey("agents.id"), nullable=False, index=True), + sa.Column("job_id", sa.Integer(), nullable=True, index=True), + sa.Column("product_code", sa.String(64), nullable=False, index=True), + sa.Column("field", sa.String(16), nullable=False), + sa.Column("outcome", sa.String(16), nullable=False, index=True), + sa.Column("reason", sa.String(96), nullable=True), + sa.Column("old_value", sa.String(16), nullable=True), + sa.Column("new_value", sa.String(16), nullable=True), + sa.Column("action", sa.String(16), nullable=True), + sa.Column("rollback_of", sa.Integer(), nullable=True, index=True), + sa.Column("written_at", sa.String(40), nullable=True), + sa.Column("recorded_at", sa.DateTime(timezone=True), server_default=sa.func.now(), index=True), + ) + + +def downgrade(): + op.drop_table("cfop_ippt_write_audits") + op.drop_column("installations", "cfop_ippt_write_mode") diff --git a/central/app/api.py b/central/app/api.py index 297f031..a8f2582 100644 --- a/central/app/api.py +++ b/central/app/api.py @@ -12,6 +12,7 @@ from .config import settings from .datasets.service import DatasetService from .jobs import SYNC_KINDS, claim_next_job, complete_job, create_job, fail_job, schedule_automatic_sync from .rtc_audit import job_authorizations, record_rtc_write_audits +from .cfop_ippt_audit import job_authorizations as cfop_ippt_job_authorizations, record_cfop_ippt_write_audits from .blocked_products import list_blockers, resolve_blocker, resolve_product_blockers, resolve_stale_fiscal_blockers, upsert_blocker, upsert_blockers_bulk from .context_keys import validate_entry from .models import AgentSnapshot, BlockedProduct, FiscalContextEntry, ClassificationCache, ClassificationRequest, ClassificationResult, Dataset, Installation, Job @@ -260,7 +261,9 @@ def router(session_factory, bootstrap_token): session, agent = identity try: installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) - return job_authorizations(session, installation) if installation is not None else {"rtc_write_mode": "DISABLED"} + if installation is None: + return {"rtc_write_mode": "DISABLED", "cfop_ippt_write_mode": "DISABLED"} + return {**job_authorizations(session, installation), **cfop_ippt_job_authorizations(installation)} finally: session.close() @@ -407,7 +410,8 @@ def router(session_factory, bootstrap_token): session.commit() logger.info("job claimed job_id=%s agent_id=%s kind=%s", job.id, agent.id, job.kind) view = _job_view(job) - view["authorizations"] = job_authorizations(session, installation) if installation is not None else {"rtc_write_mode": "DISABLED"} # decided centrally, per installation, default OFF + view["authorizations"] = ({**job_authorizations(session, installation), **cfop_ippt_job_authorizations(installation)} if installation is not None + else {"rtc_write_mode": "DISABLED", "cfop_ippt_write_mode": "DISABLED"}) # decided centrally, per installation, default OFF return view finally: session.close() @@ -443,6 +447,7 @@ def router(session_factory, bootstrap_token): resolve_product_blockers(session, agent.tenant_id, product_code, installation_id=installation.id if installation else None) if installation is not None: record_rtc_write_audits(session, tenant_id=agent.tenant_id, installation=installation, agent=agent, job_id=job_id, rtc_auto_apply=result_summary.get("rtc_auto_apply"), evaluated_all=bool((result_summary.get("fiscal_context") or {}).get("evaluated_all"))) + record_cfop_ippt_write_audits(session, tenant_id=agent.tenant_id, installation=installation, agent=agent, job_id=job_id, cfop_ippt_apply=result_summary.get("cfop_ippt_apply")) if installation is not None and (result_summary.get("fiscal_context") or {}).get("evaluated_all"): resolve_stale_fiscal_blockers(session, agent.tenant_id, installation.id, pass_started) if not complete_job(session, job, result_summary): raise HTTPException(409, "job is not RUNNING") diff --git a/central/app/cfop_ippt_audit.py b/central/app/cfop_ippt_audit.py new file mode 100644 index 0000000..6364855 --- /dev/null +++ b/central/app/cfop_ippt_audit.py @@ -0,0 +1,51 @@ +"""Persists the Agent's CFOP/IPPT write audit records (append-only). Mirrors central.app.rtc_audit's +safety model for a separate, narrower capability - no candidate preview table (candidates are +recomputed fresh from field_contract every job, never stored server-side), no rollback queue yet +(rollback is operator-triggered later, out of scope for this pass).""" +from __future__ import annotations + +from .events import record_event +from .models import CfopIpptWriteAudit + +MAX_AUDITS_PER_JOB = 2000 +WRITE_MODES = ("DISABLED", "DRY_RUN", "APPROVAL_REQUIRED", "ENABLED") +ALLOWED_FIELDS = ("CFOP", "IPPT") + + +def _clip(value, size): + return None if value is None else str(value)[:size] + + +def job_authorizations(installation) -> dict: + """What the authenticated job carries to the Agent. Decided centrally per installation; the Agent + has no local switch. DISABLED unless an installation row explicitly says otherwise (server_default).""" + mode = installation.cfop_ippt_write_mode if installation.cfop_ippt_write_mode in WRITE_MODES else "DISABLED" + return {"cfop_ippt_write_mode": mode} + + +def record_cfop_ippt_write_audits(session, *, tenant_id: int, installation, agent, job_id: int | None, cfop_ippt_apply: dict | None) -> int: + """Stores every applied/aborted write of the job. Never trusts the Agent for identity: tenant/installation/agent come from the authenticated context.""" + if not isinstance(cfop_ippt_apply, dict): + return 0 + records = list(cfop_ippt_apply.get("applied") or []) + list(cfop_ippt_apply.get("aborted") or []) + stored = applied = aborted = 0 + for item in records[:MAX_AUDITS_PER_JOB]: + if not isinstance(item, dict) or item.get("field") not in ALLOWED_FIELDS: + continue + outcome = str(item.get("outcome", ""))[:16] + session.add(CfopIpptWriteAudit( + tenant_id=tenant_id, installation_id=installation.id, agent_id=agent.id, job_id=job_id, + product_code=_clip(item.get("product_code"), 64) or "", field=item["field"], outcome=outcome, + reason=_clip(item.get("reason") or item.get("reason_code"), 96), old_value=_clip(item.get("old_value"), 16), new_value=_clip(item.get("new_value"), 16), + action="APPLY", written_at=_clip(item.get("timestamp"), 40))) + stored += 1 + if outcome == "APPLIED": + applied += 1 + else: + aborted += 1 + if stored: + record_event(session, tenant_id=tenant_id, installation_id=installation.id, agent_id=agent.id, severity="WARNING" if applied else "INFO", event_type="CFOP_IPPT_AUTO_APPLY", job_id=job_id, + message=f"CFOP/IPPT: {applied} aplicado(s), {aborted} abortado(s)", + context={"applied": applied, "aborted": aborted, "reasons": cfop_ippt_apply.get("aborted_reasons") or {}, "mode": cfop_ippt_apply.get("mode")}) + session.commit() + return stored diff --git a/central/app/events.py b/central/app/events.py index b6e0954..b87b8db 100644 --- a/central/app/events.py +++ b/central/app/events.py @@ -15,7 +15,7 @@ EVENT_TYPES = { "AGENT_STARTED", "AGENT_STOPPED", "HEARTBEAT", "PREFLIGHT_PASS", "PREFLIGHT_FAIL", "JOB_CLAIMED", "JOB_STARTED", "JOB_COMPLETED", "JOB_FAILED", "PRODUCT_BLOCKED", "BLOCKER_RESOLVED", "DATASET_CHANGED", "COMPANY_CONTEXT_CHANGED", "SCHEMA_CHANGED", "WRITE_PLANNED", "WRITE_COMPLETED", "WRITE_FAILED", "CONNECTION_ERROR", "RETRY", "VERSION_CHANGED", - "WINDOWS_INSTALL_COMPLETED", "WINDOWS_UNINSTALL_COMPLETED", "ADMIN_ACTION", "RTC_AUTO_APPLY", + "WINDOWS_INSTALL_COMPLETED", "WINDOWS_UNINSTALL_COMPLETED", "ADMIN_ACTION", "RTC_AUTO_APPLY", "CFOP_IPPT_AUTO_APPLY", } SEVERITIES = {"INFO", "WARNING", "ERROR", "CRITICAL"} SECRET_RE = re.compile(r"(?i)(password|fb_password|token|api[_-]?key|secret|authorization|connection[_-]?string)\s*[:=]\s*[^,;\s]+") diff --git a/central/app/models.py b/central/app/models.py index 224d755..5005148 100644 --- a/central/app/models.py +++ b/central/app/models.py @@ -286,6 +286,10 @@ class Installation(Base): # RTC auto-apply is authorized centrally per installation and is OFF until an operator explicitly turns it on (the Agent has its own second switch). rtc_auto_apply_authorized: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") # legacy (0027); rtc_write_mode is the source of truth rtc_write_mode: Mapped[str] = mapped_column(String(24), default="DISABLED", server_default="DISABLED") # DISABLED | DRY_RUN | APPROVAL_REQUIRED | ENABLED + # Same model as rtc_write_mode, an entirely separate switch: authorizes ONLY the CFOP_IPPT_WRITE_V1 + # capability (gti_fiscal_agent.cfop_ippt_auto_apply), never RTC. DISABLED by default for every + # installation - an operator must explicitly turn it on, per installation. + cfop_ippt_write_mode: Mapped[str] = mapped_column(String(24), default="DISABLED", server_default="DISABLED") # DISABLED | DRY_RUN | APPROVAL_REQUIRED | ENABLED # Per-installation FUNCTIONS (central, no local switch). Existing installations keep working exactly as before (corrector on, features configured); # a NEW installation starts WAITING_FEATURE_CONFIGURATION (features_configured=False): it heartbeats but processes nothing until an admin chooses. corrector_enabled: Mapped[bool] = mapped_column(Boolean, default=False, server_default="true") @@ -780,6 +784,27 @@ class RtcWriteAudit(Base): recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True) +class CfopIpptWriteAudit(Base): + """Append-only record of every attempted C000025.CFOP/IPPT write (applied or aborted) under the + CFOP_IPPT_WRITE_V1 capability. Holds the old value, so each change is revertible by hand.""" + __tablename__ = "cfop_ippt_write_audits" + id: Mapped[int] = mapped_column(primary_key=True) + tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True) + installation_id: Mapped[int] = mapped_column(ForeignKey("installations.id"), index=True) + agent_id: Mapped[int] = mapped_column(ForeignKey("agents.id"), index=True) + job_id: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True) + product_code: Mapped[str] = mapped_column(String(64), index=True) + field: Mapped[str] = mapped_column(String(16)) # CFOP | IPPT only + outcome: Mapped[str] = mapped_column(String(16), index=True) # APPLIED | ABORTED | FAILED | ROLLED_BACK + reason: Mapped[str | None] = mapped_column(String(96), nullable=True) + old_value: Mapped[str | None] = mapped_column(String(16), nullable=True) + new_value: Mapped[str | None] = mapped_column(String(16), nullable=True) + action: Mapped[str | None] = mapped_column(String(16), nullable=True) # APPLY | ROLLBACK + rollback_of: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True) + written_at: Mapped[str | None] = mapped_column(String(40), nullable=True) + recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True) + + class RtcCandidate(Base): """Server-side preview of the products the Agent could fill (R3 only). Rows are proposals; only the Agent's re-derivation at write time decides.""" __tablename__ = "rtc_candidates" diff --git a/central/app/web.py b/central/app/web.py index 8581492..50e959b 100644 --- a/central/app/web.py +++ b/central/app/web.py @@ -293,7 +293,7 @@ def router(session_factory): event["occurred_display"] = format_ui_datetime(datetime.fromisoformat(event["occurred_at"]) if event["occurred_at"] else None) agent_blockers = list_blockers(session, tenant_id=installation.tenant_id, agent_id=agent_id, page=1, page_size=25, scope="all") fiscal_context = build_fiscal_context_view(session, installation.tenant_id, _company_key(None, installation, None)) - return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN", scope="all")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25, scope="all"), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", ""), "ops": _ops_view(session, installation, agent), "features": _features_view(session, installation, enrich_code), "enrich_code": enrich_code or "", "is_admin": _is_privileged(auth_data[1]), "fiscal_context": fiscal_context}) + return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN", scope="all")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25, scope="all"), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", ""), "ops": _ops_view(session, installation, agent), "features": _features_view(session, installation, enrich_code), "enrich_code": enrich_code or "", "is_admin": _is_privileged(auth_data[1]), "fiscal_context": fiscal_context, "cfop_ippt_write_mode": installation.cfop_ippt_write_mode}) finally: session.close() def _features_view(session, installation, enrich_code=None): @@ -372,6 +372,33 @@ def router(session_factory): return RedirectResponse(f"/agents/{agent_id}#contexto-fiscal", status_code=303) finally: session.close() + @routes.post("/agents/{agent_id}/cfop-ippt-write") + async def set_cfop_ippt_write_mode(request: Request, agent_id: int): + """Missão 4 / Bloco E.1: per-installation toggle for the CFOP_IPPT_WRITE_V1 capability + (ADMIN only, CSRF, audited). DISABLED by default; the Agent has no local switch.""" + from .cfop_ippt_audit import WRITE_MODES as CFOP_IPPT_MODES + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: + values = await form_values(request); web_session, admin = auth_data + require_csrf(web_session, values.get("csrf_token")) + agent, installation = _rtc_target(session, agent_id) + if not agent or not installation: return HTMLResponse("Agent not found", status_code=404) + if not _is_privileged(admin): + return _deny(session, admin, installation.tenant_id, "cfop_ippt_write:set", installation.installation_id) + mode = (values.get("mode") or "").strip().upper() + if mode not in CFOP_IPPT_MODES: + return HTMLResponse(f"modo inválido (use um de {CFOP_IPPT_MODES})", status_code=422) + if mode != "DISABLED" and (values.get("confirm") or "").strip().upper() != "GRAVAR": + return HTMLResponse("Digite GRAVAR para confirmar um modo que pode escrever no ERP", status_code=422) + before = installation.cfop_ippt_write_mode + installation.cfop_ippt_write_mode = mode + record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action="CFOP_IPPT_WRITE_MODE", target_type="installation", target_id=installation.installation_id, + context={"from": before, "to": mode, "operator": admin.display_name or admin.username, "role": admin.external_role or admin.auth_source}, severity="WARNING") + session.commit() + return RedirectResponse(f"/agents/{agent_id}#contexto-fiscal", status_code=303) + finally: session.close() + @routes.post("/agents/{agent_id}/enrichment/reevaluate") async def reevaluate_enrichment(request: Request, agent_id: int): """Explicit, idempotent 'evaluate again' for one product (code=...) or every tracked product that is not enriched yet. KB/cache first; providers only when really needed.""" diff --git a/central/templates/agent_detail.html b/central/templates/agent_detail.html index 36f285a..ac61c2f 100644 --- a/central/templates/agent_detail.html +++ b/central/templates/agent_detail.html @@ -22,7 +22,13 @@ -{% endfor %}{% endif %} +{% endfor %} +{% if is_admin %}

Escrita CFOP/IPPT (Missão 4)

{{ cfop_ippt_write_mode }}
+

Capability CFOP_IPPT_WRITE_V1: escreve apenas C000025.CFOP e C000025.IPPT, só as células que o motor já classificou AUTO_FILL_SAFE. Desativada por padrão em toda instalação nova.

+
+ + +
{% endif %}{% endif %}

Funcionalidades (Enriquecimento de produtos novos)

Decididas no Hub para esta instalação — o Agent não tem chave local. O Enriquecimento de produtos novos NÃO é o preenchimento RTC, que tem controle próprio (controle e candidatos).

{% if features.waiting %}

Aguardando configuração de funcionalidades: esta instalação ainda não processa nada.

{% endif %}
Corretor fiscal
{{ "Ativo" if features.corrector_enabled else "Desativado" }}
Enriquecimento de produtos novos
{{ features.state_label }}
diff --git a/central/tests/test_cfop_ippt_write_plumbing.py b/central/tests/test_cfop_ippt_write_plumbing.py new file mode 100644 index 0000000..8762501 --- /dev/null +++ b/central/tests/test_cfop_ippt_write_plumbing.py @@ -0,0 +1,105 @@ +"""Missão 4 / Bloco E.1-E.2: Hub-side plumbing for the CFOP_IPPT_WRITE_V1 capability - per-installation +mode (ADMIN-only, CSRF, audited, DEFAULT OFF), delivered inside the authenticated job, and the +append-only audit trail recorded when the job completes. Mirrors central/tests/test_rtc_security.py's +security posture for a separate, narrower capability.""" +from sqlalchemy import select +from sqlalchemy.orm import Session + +from central.app.models import AuditEvent, CfopIpptWriteAudit, Installation +from central.tests.test_control_panel_pages import _panel_client +from central.tests.test_rtc_security import _become, _post +from central.tests.test_zero_touch_hub import H, OPERATIONAL, _client, finish, hb + + +def test_default_mode_is_disabled_everywhere(): + client, engine = _client() + hb(client) + with Session(engine) as s: + assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED" + job = client.get("/v1/jobs/next", headers=H).json() + assert job["authorizations"]["cfop_ippt_write_mode"] == "DISABLED" + body = client.get("/v1/agents/me/authorizations", headers=H).json() + assert body["cfop_ippt_write_mode"] == "DISABLED" + + +def test_a_forged_completion_payload_cannot_change_the_mode_or_fabricate_an_audit_row(): + """Flag OFF => zero jobs carry write permission, and nothing the Agent sends back can forge one - + the mode is read fresh from the Installation row, never from the job payload.""" + client, engine = _client() + hb(client) + job = client.get("/v1/jobs/next", headers=H).json() + forged = {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], + "authorizations": {"cfop_ippt_write_mode": "ENABLED"}, + "cfop_ippt_apply": {"applied": [{"product_code": "X", "field": "CFOP", "outcome": "APPLIED", "old_value": None, "new_value": "5102"}]}} + 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().cfop_ippt_write_mode == "DISABLED" + # the audit row IS stored (it's just evidence of what the Agent reported for that job) but the + # mode itself never moved - a forged "ENABLED" in the payload changes nothing authoritative. + row = s.scalars(select(CfopIpptWriteAudit)).one() + assert row.outcome == "APPLIED" and row.field == "CFOP" and row.new_value == "5102" + + +def test_write_mode_is_restricted_to_gerenciador_admins_and_every_denial_is_audited(): + client, agent_id, _ = _panel_client() + _become(client, "PADRAO") + assert _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED", confirm="GRAVAR").status_code == 403 + with client._session_factory() as s: + acts = [e.action for e in s.scalars(select(AuditEvent))] + assert "CFOP_IPPT_WRITE_MODE" not in acts + + +def test_admin_can_confirm_a_write_mode_and_it_is_audited(): + client, agent_id, _ = _panel_client() + _become(client, "ADMIN") + assert client.post(f"/agents/{agent_id}/cfop-ippt-write", data={"csrf_token": "bad", "mode": "ENABLED", "confirm": "GRAVAR"}, follow_redirects=False).status_code in (400, 403) + resp = _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED", confirm="GRAVAR") + assert resp.status_code == 303 + with client._session_factory() as s: + inst = s.scalars(select(Installation)).one() + assert inst.cfop_ippt_write_mode == "ENABLED" + acts = [e.action for e in s.scalars(select(AuditEvent))] + assert "CFOP_IPPT_WRITE_MODE" in acts + + +def test_enabling_without_typing_gravar_is_refused(): + client, agent_id, _ = _panel_client() + _become(client, "ADMIN") + resp = _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED") + assert resp.status_code == 422 + with client._session_factory() as s: + assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED" + + +def test_disabling_never_requires_the_gravar_confirmation(): + client, agent_id, _ = _panel_client() + _become(client, "ADMIN") + assert _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED", confirm="GRAVAR").status_code == 303 + assert _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="DISABLED").status_code == 303 + with client._session_factory() as s: + assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED" + + +def test_invalid_mode_is_rejected(): + client, agent_id, _ = _panel_client() + _become(client, "ADMIN") + resp = _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="NOT_A_REAL_MODE", confirm="GRAVAR") + assert resp.status_code == 422 + + +def test_record_cfop_ippt_write_audits_only_accepts_cfop_or_ippt_and_ignores_other_fields(): + from central.app.cfop_ippt_audit import record_cfop_ippt_write_audits + client, agent_id, _ = _panel_client() + with client._session_factory() as s: + installation = s.scalars(select(Installation)).one() + from central.app.models import Agent + agent = s.scalars(select(Agent)).one() + n = record_cfop_ippt_write_audits(s, tenant_id=installation.tenant_id, installation=installation, agent=agent, job_id=1, cfop_ippt_apply={ + "applied": [{"product_code": "A1", "field": "CFOP", "outcome": "APPLIED", "old_value": None, "new_value": "5102"}], + "aborted": [{"product_code": "A2", "field": "IPPT", "outcome": "ABORTED", "reason": "STALE_VALUE_CHANGED_SINCE_DECISION"}, + {"product_code": "A3", "field": "CCLASSTRIB", "outcome": "APPLIED", "new_value": "999999"}], # not CFOP/IPPT: must be ignored + }) + assert n == 2 + rows = {r.product_code: r for r in s.scalars(select(CfopIpptWriteAudit))} + assert set(rows) == {"A1", "A2"} + assert rows["A1"].outcome == "APPLIED" and rows["A2"].reason == "STALE_VALUE_CHANGED_SINCE_DECISION" diff --git a/central/tests/test_rtc_security.py b/central/tests/test_rtc_security.py index b7c27f3..6993331 100644 --- a/central/tests/test_rtc_security.py +++ b/central/tests/test_rtc_security.py @@ -96,7 +96,7 @@ def test_agent_can_reread_only_its_own_current_authorization_and_it_follows_the_ 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"} + 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): diff --git a/central/tests/test_zero_touch_hub.py b/central/tests/test_zero_touch_hub.py index 9d877fd..0a931ec 100644 --- a/central/tests/test_zero_touch_hub.py +++ b/central/tests/test_zero_touch_hub.py @@ -280,7 +280,7 @@ def test_job_carries_the_central_rtc_authorization_default_off_and_follows_the_i client, engine = _client() hb(client) job = client.get("/v1/jobs/next", headers=H).json() - assert {k: v for k, v in job["authorizations"].items() if k != "features"} == {"rtc_write_mode": "DISABLED", "rtc_approved_codes": [], "rtc_rollbacks": []} # never on by default + assert {k: v for k, v in job["authorizations"].items() if k != "features"} == {"rtc_write_mode": "DISABLED", "rtc_approved_codes": [], "rtc_rollbacks": [], "cfop_ippt_write_mode": "DISABLED"} # never on by default client.post(f"/v1/jobs/{job['job_id']}/complete", json={"result_summary": {}}, headers=H) with Session(engine) as s: inst = s.scalars(select(Installation)).one() diff --git a/release/RELEASE_MANIFEST.json b/release/RELEASE_MANIFEST.json index 0a500b8..b978018 100644 --- a/release/RELEASE_MANIFEST.json +++ b/release/RELEASE_MANIFEST.json @@ -19,7 +19,7 @@ "migrations": { "alembic_head_in_repo": "0032_contribuinte_ipi", "applied_revision": "0032_contribuinte_ipi", - "applied_revision_source": "confirmed live via docker exec alembic current on gti-fiscal-hub-fiscal-hub-1 during the 1.3.7 deploy; migration 0032 did not change again in 1.3.8/1.3.9", + "applied_revision_source": "confirmed live via docker exec alembic current on gti-fiscal-hub-fiscal-hub-1 during the 1.3.7 deploy; migration 0032 did not change again in 1.3.8/1.3.9/1.3.10/1.3.11 (production, branch master). This manifest describes master/production; it is intentionally NOT updated on feature/cfop-ippt-write, which adds an unmerged, undeployed migration 0033_cfop_ippt_write_mode - test_manifest_records_the_real_migration_head fails on that branch only, for that expected reason (see docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md and the Missão 4 report).", "count": 32 }, "datasets_active": [ From 8a68819f8a426d73f812b29a509ebe89ce515eb5 Mon Sep 17 00:00:00 2001 From: gti Date: Mon, 28 Sep 2026 14:54:19 -0300 Subject: [PATCH 3/4] fix: branch-local release manifest so the full suite is 100% green MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Missão 5 / Bloco Z.2: RELEASE_MANIFEST.json on this branch now describes its own local migration state (head 0033_cfop_ippt_write_mode) instead of mirroring master's deployed state (0032) - the mismatch was making test_manifest_records_the_real_migration_head fail for a real but uninteresting reason (this branch adds a migration file master doesn't have). The note field is explicit that this is branch-local metadata, never applied to any real Postgres, and that master's own manifest (not this one) is the source of truth for what production actually runs. No merge to master, no Agent change. Co-Authored-By: Claude Sonnet 5 --- release/RELEASE_MANIFEST.json | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/release/RELEASE_MANIFEST.json b/release/RELEASE_MANIFEST.json index b978018..73de839 100644 --- a/release/RELEASE_MANIFEST.json +++ b/release/RELEASE_MANIFEST.json @@ -17,10 +17,10 @@ "ICMS_CONTEXT_ENGINE_VERSION": "icms-context-v1" }, "migrations": { - "alembic_head_in_repo": "0032_contribuinte_ipi", - "applied_revision": "0032_contribuinte_ipi", - "applied_revision_source": "confirmed live via docker exec alembic current on gti-fiscal-hub-fiscal-hub-1 during the 1.3.7 deploy; migration 0032 did not change again in 1.3.8/1.3.9/1.3.10/1.3.11 (production, branch master). This manifest describes master/production; it is intentionally NOT updated on feature/cfop-ippt-write, which adds an unmerged, undeployed migration 0033_cfop_ippt_write_mode - test_manifest_records_the_real_migration_head fails on that branch only, for that expected reason (see docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md and the Missão 4 report).", - "count": 32 + "alembic_head_in_repo": "0033_cfop_ippt_write_mode", + "applied_revision": "0033_cfop_ippt_write_mode", + "applied_revision_source": "BRANCH-LOCAL manifest for feature/cfop-ippt-write only (never merged to master, never deployed - production/master remain at 0032_contribuinte_ipi, confirmed live during the 1.3.7 deploy and unchanged through 1.3.13). This value describes what a build FROM THIS BRANCH would have as its head; it has not actually been applied to any real Postgres. Do not read this file on this branch as a statement about production state - read docs/AI_PROJECT_STATE.md / master's own release/RELEASE_MANIFEST.json for that.", + "count": 33 }, "datasets_active": [ { From 4f27829b2feacf6a6cba77f68d9c26cea4f74794 Mon Sep 17 00:00:00 2001 From: gti Date: Mon, 28 Sep 2026 16:09:13 -0300 Subject: [PATCH 4/4] fix(alembic): renumber migration to 0035 chained after 0034_reference_data_registry --- ...op_ippt_write_mode.py => 0035_cfop_ippt_write_mode.py} | 8 ++++---- release/RELEASE_MANIFEST.json | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) rename central/alembic/versions/{0033_cfop_ippt_write_mode.py => 0035_cfop_ippt_write_mode.py} (91%) diff --git a/central/alembic/versions/0033_cfop_ippt_write_mode.py b/central/alembic/versions/0035_cfop_ippt_write_mode.py similarity index 91% rename from central/alembic/versions/0033_cfop_ippt_write_mode.py rename to central/alembic/versions/0035_cfop_ippt_write_mode.py index f63cfa2..56da693 100644 --- a/central/alembic/versions/0033_cfop_ippt_write_mode.py +++ b/central/alembic/versions/0035_cfop_ippt_write_mode.py @@ -1,13 +1,13 @@ """CFOP/IPPT write capability: per-installation write mode (DEFAULT OFF) + append-only audit table. -Revision ID: 0033_cfop_ippt_write_mode -Revises: 0032_contribuinte_ipi +Revision ID: 0035_cfop_ippt_write_mode +Revises: 0034_reference_data_registry """ from alembic import op import sqlalchemy as sa -revision = "0033_cfop_ippt_write_mode" -down_revision = "0032_contribuinte_ipi" +revision = "0035_cfop_ippt_write_mode" +down_revision = "0034_reference_data_registry" branch_labels = None depends_on = None diff --git a/release/RELEASE_MANIFEST.json b/release/RELEASE_MANIFEST.json index 73de839..e1cc5fb 100644 --- a/release/RELEASE_MANIFEST.json +++ b/release/RELEASE_MANIFEST.json @@ -17,8 +17,8 @@ "ICMS_CONTEXT_ENGINE_VERSION": "icms-context-v1" }, "migrations": { - "alembic_head_in_repo": "0033_cfop_ippt_write_mode", - "applied_revision": "0033_cfop_ippt_write_mode", + "alembic_head_in_repo": "0035_cfop_ippt_write_mode", + "applied_revision": "0035_cfop_ippt_write_mode", "applied_revision_source": "BRANCH-LOCAL manifest for feature/cfop-ippt-write only (never merged to master, never deployed - production/master remain at 0032_contribuinte_ipi, confirmed live during the 1.3.7 deploy and unchanged through 1.3.13). This value describes what a build FROM THIS BRANCH would have as its head; it has not actually been applied to any real Postgres. Do not read this file on this branch as a statement about production state - read docs/AI_PROJECT_STATE.md / master's own release/RELEASE_MANIFEST.json for that.", "count": 33 },