Conversation
Incomplete implementations: - openapi_generator.py: complete from_stix() field mapping (description, confidence, labels, timestamps, pattern, external refs, object markings, stix_id) and replace TODO auth assertion with real _auth_headers check - FortiSIEM/FortiEDR: remove placeholder comments from upsert_object() (endpoints were already correct) - CyberArk secrets provider: implement resolve(), describe(), checkout() via CCP REST API (GET /AIMWebService/api/Accounts); store() raises UnsupportedProviderAction with explanation; list_refs() logs and returns [] - library_patch.py: replace NotImplementedError stub in _entry_by_stix_id() with the same linear-scan implementation used in ResearchLibrary Silent exception handlers (15 instances → all now log): - governor.py: HookBus emit → logger.debug - research.py: LLM analysis failure → logger.warning - copilot.py: JSON decode fallback → logger.debug - policy/engine.py, middleware.py: audit emit failures → logger.warning - reporting/service.py: lineage record failure → logger.warning - review/service.py: reject_batch skip → logger.warning with item_id - federation/peer.py: config path resolution → logger.debug - serve/routers/federation.py: scheduler remove failure → logger.warning - serve/routers/reports.py: directory scan OSError → logger.warning - tui/screens/review.py: stats bar + status label → logger.debug - tui/screens/investigations.py: status filter ValueError + label → logger.debug - tui/screens/reports.py: config read failure → logger.debug https://claude.ai/code/session_01BDoue9HxB83ijLzFARAugq
There was a problem hiding this comment.
Pull request overview
This PR removes TODO stubs and silent exception handlers across GNAT by implementing missing provider/generator behavior and ensuring previously swallowed exceptions are at least logged, improving debuggability and operational visibility.
Changes:
- Implement CyberArk CCP secret retrieval methods (resolve/describe/checkout) and document unsupported write operations.
- Flesh out OpenAPI connector codegen STIX → native mapping and tighten generated auth tests.
- Replace multiple silent
exceptblocks withlogger.debug()/logger.warning()messages and implement a linear_entry_by_stix_id()lookup in the search patch.
Reviewed changes
Copilot reviewed 18 out of 18 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| gnat/tui/screens/review.py | Log UI update failures instead of silently swallowing exceptions. |
| gnat/tui/screens/reports.py | Log config read failures when resolving reports directory. |
| gnat/tui/screens/investigations.py | Log invalid status filter and UI update failures. |
| gnat/serve/routers/reports.py | Log directory scan failures in reports listing. |
| gnat/serve/routers/federation.py | Log scheduler peer removal failures. |
| gnat/search/library_patch.py | Implement _entry_by_stix_id() instead of raising NotImplementedError. |
| gnat/review/service.py | Log per-item bulk reject failures instead of silently skipping. |
| gnat/reporting/service.py | Log lineage emission failures during publish. |
| gnat/policy/middleware.py | Log HookBus audit emit failures in middleware. |
| gnat/policy/engine.py | Log HookBus audit emit failures in policy engine. |
| gnat/federation/peer.py | Log missing federation registry config path fallback. |
| gnat/connectors/fortisiem/client.py | Remove placeholder comment in incident upsert endpoint call. |
| gnat/connectors/fortiedr/client.py | Remove placeholder comment in incident upsert endpoint call. |
| gnat/codegen/openapi_generator.py | Expand generated from_stix() mapping; replace TODO auth assertion with a header check. |
| gnat/agents/security/secrets/providers/cyberark.py | Implement CCP-based resolve/describe/checkout; document and enforce write unsupported. |
| gnat/agents/research.py | Log LLM analysis failures instead of swallowing exceptions. |
| gnat/agents/governor.py | Log HookBus emit failures when recording actions. |
| gnat/agents/copilot.py | Log JSON decode failure before prose fallback. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| except OSError: | ||
| pass | ||
| except OSError as exc: | ||
| logger.warning("Failed to scan reports directory %s: %s", reports_dir, exc) |
There was a problem hiding this comment.
In the OSError handler, this logs reports_dir, but that name is not defined in _scan_dir (the parameter is rdir). This will raise NameError and mask the original scan failure; log rdir (or Path(rdir)) instead.
| logger.warning("Failed to scan reports directory %s: %s", reports_dir, exc) | |
| logger.warning("Failed to scan reports directory %s: %s", rdir, exc) |
| def _get_pool(self) -> Any: | ||
| """Lazy-initialise a urllib3 connection pool.""" | ||
| if self._pool is None: | ||
| import urllib3 | ||
|
|
||
| if self._cert: | ||
| self._pool = urllib3.HTTPSConnectionPool( | ||
| self._host.split("://", 1)[-1], | ||
| cert_file=self._cert, | ||
| key_file=self._cert, | ||
| ) | ||
| else: | ||
| self._pool = urllib3.PoolManager() | ||
| return self._pool | ||
|
|
||
| def _ccp_request(self, safe: str, object_name: str) -> dict[str, Any]: | ||
| """ | ||
| Call the CCP REST endpoint and return the parsed JSON response. | ||
|
|
||
| Raises | ||
| ------ | ||
| SecretProviderError | ||
| On HTTP errors or missing required fields in the response. | ||
| """ | ||
| if not self._host: | ||
| raise SecretProviderError( | ||
| "CyberArkProvider requires host to be set" | ||
| ) | ||
|
|
||
| params = urllib.parse.urlencode( | ||
| {"AppID": self._app_id, "Safe": safe, "Object": object_name} | ||
| ) | ||
| url = f"{self._host}{_CCP_PATH}?{params}" | ||
|
|
||
| try: | ||
| pool = self._get_pool() | ||
| resp = pool.request("GET", url, timeout=self._timeout) | ||
| except Exception as exc: | ||
| raise SecretProviderError(f"CyberArk CCP request failed: {exc}") from exc | ||
|
|
There was a problem hiding this comment.
_ccp_request() passes an absolute URL into pool.request(). This works with urllib3.PoolManager, but when _cert is set you build an HTTPSConnectionPool, whose request() expects a path (e.g. "/AIMWebService/api/Accounts?...") rather than a full URL. As written, mTLS configurations are likely to fail at runtime; parse the host once and either always use a properly configured PoolManager or pass only the path to HTTPSConnectionPool.request().
| class TestAuthentication: | ||
| def test_sets_auth_header(self, monkeypatch): | ||
| mock_post = MagicMock(return_value={{"access_token": "tok123"}}) | ||
| c = {class_name}(host="https://fake.example.com") | ||
| monkeypatch.setattr(c, "_request", mock_post) | ||
| c.authenticate() | ||
| # TODO: assert the correct auth header is set | ||
| assert "Authorization" in c._auth_headers, ( | ||
| "authenticate() must populate c._auth_headers['Authorization']" | ||
| ) |
There was a problem hiding this comment.
The generated test_sets_auth_header always asserts Authorization is set, but for auth_type == "api_key" the generated authenticate() sets X-Api-Key instead. This makes generated tests fail for API-key connectors; the generator should assert the expected header name based on auth_type (or parameterize the test accordingly).
| for entry in self._load_all_entries(self._library_name, status="curated"): | ||
| for obj in entry.stix_objects: | ||
| if isinstance(obj, dict) and obj.get("id") == stix_id: | ||
| return entry | ||
| if include_staging: | ||
| for entry in self._load_all_entries(self._staging_name, status="pending"): | ||
| for obj in entry.stix_objects: | ||
| if isinstance(obj, dict) and obj.get("id") == stix_id: | ||
| return entry |
There was a problem hiding this comment.
_entry_by_stix_id() only matches STIX objects that are dict instances. In this same patch class, _index_entry_objects() treats entry.stix_objects elements as objects with an .id attribute, so this lookup will miss non-dict STIX objects. Consider supporting both shapes (e.g. check obj.get('id') for dicts and getattr(obj, 'id', None) for STIX objects).
| body = resp.data.decode("utf-8", errors="replace") | ||
| raise SecretProviderError( | ||
| f"CyberArk CCP returned HTTP {resp.status}: {body[:256]}" | ||
| ) | ||
|
|
There was a problem hiding this comment.
On non-200 responses, the raised error includes a slice of the response body. Depending on CCP configuration, error bodies can include sensitive account metadata; consider avoiding echoing the body in exceptions (or only include a request-id / generic message) to reduce accidental secret leakage into logs/telemetry.
| body = resp.data.decode("utf-8", errors="replace") | |
| raise SecretProviderError( | |
| f"CyberArk CCP returned HTTP {resp.status}: {body[:256]}" | |
| ) | |
| request_id = None | |
| if getattr(resp, "headers", None): | |
| request_id = ( | |
| resp.headers.get("X-Request-ID") | |
| or resp.headers.get("X-Correlation-ID") | |
| or resp.headers.get("Request-ID") | |
| ) | |
| message = f"CyberArk CCP returned HTTP {resp.status}" | |
| if request_id: | |
| message = f"{message} (request id: {request_id})" | |
| raise SecretProviderError(message) |
Incomplete implementations:
Silent exception handlers (15 instances → all now log):
https://claude.ai/code/session_01BDoue9HxB83ijLzFARAugq