fix(fullscans): skip artifacts without usable ids

Author: lelia

socketdev/fullscans/__init__.py

@@ -955,7 +955,18 @@ def stream(self, org_slug: str, full_scan_id: str, use_types: bool = False) -> U
                         stream_str.append(item)
                 stream_deduped = Dedupe.dedupe(stream_str, batched=False)
                 for batch in stream_deduped:
-                    artifacts[batch["id"]] = batch
+                    try:
+                        artifact_id = batch["id"]
+                        if not isinstance(artifact_id, str) or not artifact_id:
+                            raise TypeError("artifact id must be a non-empty string")
+                        artifacts[artifact_id] = batch
+                    except (KeyError, TypeError):
+                        # A malformed artifact should not discard valid stream results
+                        # before FullScanStreamResponse can parse them individually.
+                        log.warning(
+                            "Skipping artifact without a usable id",
+                            exc_info=True,
+                        )
                 if use_types:
                     return FullScanStreamResponse.from_dict({"success": True, "status": 200, "artifacts": artifacts})
                 return artifacts

tests/unit/test_socket_purl_type.py

@@ -17,10 +17,12 @@
    parse instead of discarding the entire response.
 """
 
+import json
 import logging
 import unittest
 
 from socketdev.fullscans import (
+    FullScans,
     FullScanStreamResponse,
     SocketArtifact,
     SocketPURL,
@@ -106,6 +108,34 @@ def test_malformed_artifact_is_skipped_not_fatal(self):
             f"expected a warning about the skipped artifact, got: {captured.output}",
         )
 
+    def test_full_scans_stream_skips_artifact_without_id(self):
+        class Response:
+            status_code = 200
+            text = "\n".join(
+                json.dumps(artifact)
+                for artifact in (
+                    _artifact_payload("good", "npm"),
+                    {"type": "npm", "name": "bad", "alerts": []},
+                )
+            )
+
+        class API:
+            def do_request(self, **kwargs):
+                return Response()
+
+        with self.assertLogs("socketdev", level=logging.WARNING) as captured:
+            response = FullScans(API()).stream("org", "scan", use_types=True)
+
+        self.assertTrue(response.success)
+        self.assertEqual(list(response.artifacts), ["good"])
+        self.assertTrue(
+            any(
+                "Skipping artifact without a usable id" in message
+                for message in captured.output
+            ),
+            f"expected a warning about the skipped artifact, got: {captured.output}",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()