Wait for sanitized IaC map publication in the CLI

doc/cli/cloud-security.md

@@ -430,11 +430,16 @@ secret values, source snippets, outputs and arbitrary attributes are omitted.
 Unknown or unsupported inputs make coverage partial; partial/failed pushes cannot
 delete prior mappings. Raw state and plans are refused by the push command/API.
 
-Raw extraction input is limited to 64 MiB; sanitized uploads to 10 MiB, 50,000
+Raw extraction input is limited to 64 MiB; sanitized uploads to 20 MiB, 100,000
 resources, depth 8 and strings of 4 KiB. Push requires `cloudsec.set` for the selected
 organization and feature availability. The API limits pushes to 30/minute per
-identity and organization. A successful response describes reconciliation and
-coverage, not deployment or remediation verification. Keep raw files local; only
+identity and organization. Push returns a receipt with a content hash and
+`processing` or `published` status. The CLI polls until publication and
+resubmits the same document if an interrupted worker becomes retryable. The SDK
+returns the receipt immediately; call `CloudSec.get_iac_map_status` with the
+document's repository, provider, workspace and source kind plus the receipt
+hash. Only `published` means the map is visible. A successful receipt is not
+deployment or remediation verification. Keep raw files local; only
 `sanitized-map.json` belongs in the upload step. No collection credential is used
 for response actions.
 

limacharlie/commands/cloudsec.py

@@ -5055,7 +5055,7 @@ def code_iac_map_extract(input_path, source_kind, repository, commit, workspace,
     with tempfile.TemporaryFile() as output:
         try:
             result = subprocess.run(args, stdin=subprocess.DEVNULL, stdout=output,
-                                    stderr=subprocess.DEVNULL, timeout=10, check=False,
+                                    stderr=subprocess.DEVNULL, timeout=120, check=False,
                                     env={"PATH": os.defpath})
             output.seek(0)
             raw = output.read(MAX_BYTES + 1)
@@ -5068,7 +5068,7 @@ def code_iac_map_extract(input_path, source_kind, repository, commit, workspace,
 
 
 @code_iac_map.command("push")
-@click.option("--input", "input_path", required=True, type=click.Path(exists=True, dir_okay=False), help="Local sanitized lc-iac-map/v1 JSON, at most 10 MiB. Raw state/plans are refused.")
+@click.option("--input", "input_path", required=True, type=click.Path(exists=True, dir_okay=False), help="Local sanitized lc-iac-map/v1 JSON, at most 20 MiB. Raw state/plans are refused.")
 @pass_context
 def code_iac_map_push(ctx, input_path) -> None:
     """Push only locally sanitized IaC JSON; requires cloudsec.set.
@@ -5083,5 +5083,49 @@ def code_iac_map_push(ctx, input_path) -> None:
             raw = validate_iac_map(source.read(MAX_BYTES + 1))
     except (OSError, ValueError):
         raise click.ClickException("invalid sanitized IaC map; run cloudsec code iac-map extract first") from None
-    result = _get_cloudsec(ctx).push_iac_map(raw)
-    _output(ctx, result)
+    cloudsec = _get_cloudsec(ctx)
+    document = json.loads(raw)
+    result = cloudsec.push_iac_map(raw)
+    receipt = result.get("result", {})
+    if receipt.get("status") != "processing":
+        _output(ctx, result)
+        return
+    digest = receipt.get("hash")
+    if not isinstance(digest, str) or len(digest) != 64:
+        raise click.ClickException("IaC map receipt is missing its hash; retry the push")
+    selectors = {
+        "repository": document["repository"]["name"],
+        "provider": document["repository"]["provider"],
+        "workspace": document["workspace"],
+        "source_kind": document["source_kind"],
+        "hash": digest,
+    }
+    deadline = time.monotonic() + 15 * 60
+    retries = 0
+    while time.monotonic() < deadline:
+        time.sleep(2)
+        status = cloudsec.get_iac_map_status(**selectors).get("status")
+        if status == "published":
+            receipt["status"] = "published"
+            _output(ctx, result)
+            return
+        if status == "retryable":
+            if retries >= 5:
+                raise click.ClickException("IaC map worker interrupted repeatedly; retry the same push")
+            result = cloudsec.push_iac_map(raw)
+            receipt = result.get("result", {})
+            if receipt.get("status") == "published":
+                _output(ctx, result)
+                return
+            if receipt.get("status") != "processing":
+                raise click.ClickException("IaC map resubmission returned an invalid receipt")
+            digest = receipt.get("hash")
+            if not isinstance(digest, str) or len(digest) != 64:
+                raise click.ClickException("IaC map resubmission is missing its hash")
+            selectors["hash"] = digest
+            retries += 1
+        elif status == "superseded":
+            raise click.ClickException("IaC map was superseded by a newer document")
+        elif status != "processing":
+            raise click.ClickException("IaC map status is unavailable; retry the same push")
+    raise click.ClickException("IaC map is still processing; retry the same push to check its status")

limacharlie/sdk/cloudsec.py

@@ -2691,10 +2691,11 @@ def push_iac_map(self, document: bytes | str) -> dict[str, Any]:
         """Push a sanitized IaC map with the organization's write authorization.
 
         Args:
-            document: Locally extracted lc-iac-map/v1 JSON, at most 10 MiB.
+            document: Locally extracted lc-iac-map/v1 JSON, at most 20 MiB.
 
         Returns:
-            dict: Reconcile counts, partial coverage, content hash and replay status.
+            dict: A receipt whose result.status is processing or published.
+            Poll get_iac_map_status before treating the map as visible.
 
         Raises:
             ValueError: If local preflight rejects the sanitized document.
@@ -2706,6 +2707,24 @@ def push_iac_map(self, document: bytes | str) -> dict[str, Any]:
             "POST", f"cloudsec/{self.oid}/code/iac-map",
             raw_body=raw, content_type="application/json")
 
+    def get_iac_map_status(
+        self, *, repository: str, provider: str, workspace: str,
+        source_kind: str, hash: str,
+    ) -> dict[str, Any]:
+        """Read one map receipt under the organization's write authorization.
+
+        A retryable status means the same sanitized document can be submitted
+        again; superseded means a newer map replaced this receipt.
+        """
+        return self._org.client.request(
+            "GET", f"cloudsec/{self.oid}/code/iac-map/status",
+            query_params=[
+                ("repository", repository), ("provider", provider),
+                ("workspace", workspace), ("source_kind", source_kind),
+                ("hash", hash),
+            ],
+        )
+
     def push_code_provenance(self, document: bytes | str | dict[str, Any]) -> dict[str, Any]:
         """Push build provenance without changing signed document bytes.
 

limacharlie/sdk/iac_map.py

@@ -3,10 +3,9 @@
 
 import json
 import re
-import time
 import unicodedata
 
-MAX_BYTES = 10 * 1024 * 1024
+MAX_BYTES = 20 * 1024 * 1024
 EXTRACT_COMMAND = "limacharlie cloudsec code iac-map extract --input terraform.json --source-kind state_identity --repository owner/repo --commit FULL_COMMIT --workspace default"
 
 
@@ -74,21 +73,20 @@ def validate_iac_map(document: bytes | str) -> bytes:
                          parse_constant=lambda _: _refuse())
     except (ValueError, UnicodeError, RecursionError):
         _refuse()
-    deadline = time.monotonic() + 2
     stack = [obj]
     nodes = 0
     while stack:
         value = stack.pop()
         nodes += 1
-        if nodes > 2_000_000 or time.monotonic() > deadline:
+        if nodes > 4_000_000:
             _refuse()
         if isinstance(value, dict):
             if len(value) > 256:
                 _refuse()
             stack.extend(value.keys())
             stack.extend(value.values())
         elif isinstance(value, list):
-            if len(value) > 50_000:
+            if len(value) > 100_000:
                 _refuse()
             stack.extend(value)
         elif isinstance(value, str):

tests/unit/test_iac_map.py

@@ -22,6 +22,53 @@ def test_shared_extractor_golden_and_server_tenant_route():
     org.client.request.assert_called_once_with("POST", "cloudsec/tenant-a/code/iac-map", raw_body=raw, content_type="application/json")
 
 
+def test_status_receipt_uses_only_scoped_selectors():
+    org = MagicMock(oid="tenant-a")
+    CloudSec(org).get_iac_map_status(
+        repository="owner/repo", provider="github", workspace="default",
+        source_kind="state_identity", hash="a" * 64,
+    )
+    org.client.request.assert_called_once_with(
+        "GET", "cloudsec/tenant-a/code/iac-map/status",
+        query_params=[
+            ("repository", "owner/repo"), ("provider", "github"),
+            ("workspace", "default"), ("source_kind", "state_identity"),
+            ("hash", "a" * 64),
+        ],
+    )
+
+
+def test_preflight_accepts_one_map_above_original_50k_bound():
+    doc = json.loads(GOLDEN.read_bytes())
+    resource = doc["resources"][0]
+    doc["resources"] = [dict(resource, address=f"google_storage_bucket.bucket{i}", identity={"name": f"bucket{i}"}) for i in range(50_001)]
+    raw = json.dumps(doc, separators=(",", ":")).encode()
+    assert len(raw) < MAX_BYTES
+    assert validate_iac_map(raw) == raw
+
+
+def test_cli_waits_for_publication_and_resubmits_after_worker_loss(tmp_path):
+    source = tmp_path / "map.json"
+    source.write_bytes(GOLDEN.read_bytes())
+    client = MagicMock()
+    client.push_iac_map.side_effect = [
+        {"result": {"status": "processing", "hash": "a" * 64}},
+        {"result": {"status": "processing", "hash": "b" * 64}},
+    ]
+    client.get_iac_map_status.side_effect = [
+        {"status": "retryable"}, {"status": "published"},
+    ]
+    with patch("limacharlie.commands.cloudsec._get_cloudsec", return_value=client), \
+         patch("limacharlie.commands.cloudsec.time.sleep"), \
+         patch("limacharlie.commands.cloudsec._output") as output:
+        response = CliRunner().invoke(code_iac_map, ["push", "--input", str(source)])
+    assert response.exit_code == 0, response.output
+    assert client.push_iac_map.call_count == 2
+    assert client.get_iac_map_status.call_count == 2
+    assert client.get_iac_map_status.call_args_list[1].kwargs["hash"] == "b" * 64
+    assert output.call_args.args[1]["result"]["status"] == "published"
+
+
 @pytest.mark.parametrize("raw", [b'{"values":{"password":"SENSITIVE_CANARY"}}',
     b'{"schema":"x","schema":"y"}', b"[" * 5000, b" " * (MAX_BYTES + 1),
     b'{"token":"SENSITIVE_CANARY"}', b'"\\ud800"'])
@@ -50,7 +97,7 @@ def test_offline_extract_does_not_authenticate_or_inherit_tokens(tmp_path):
     source.write_text('{"values":{"secret":"SENSITIVE_CANARY"}}')
     def run(args, **kwargs):
         assert kwargs["env"] == {"PATH": __import__("os").defpath}
-        assert kwargs["timeout"] == 10
+        assert kwargs["timeout"] == 120
         assert args[0] == "/trusted/iac-map-extract"
         kwargs["stdout"].write(GOLDEN.read_bytes())
         return MagicMock(returncode=0)