diff --git a/anchor/core/engine.py b/anchor/core/engine.py index ac9d34c..fba79ef 100644 --- a/anchor/core/engine.py +++ b/anchor/core/engine.py @@ -640,9 +640,16 @@ def evaluate_candidates(self, candidates: List[Dict[str, Any]], content: bytes, click.secho(f" [EVALUATOR] ALN-001 candidate at line {candidate['line']} discarded (validation markers found).", fg="green", dim=True) continue - # Specialized evaluator for Process Execution (SEC-007, FINOS-014, OWASP-002, RBI-018) + # Evaluate for static/dynamic downgrade and min_severity floor clamping is_proc_exec = any(r_id in rule_id for r_id in ["SEC-007", "FINOS-014", "OWASP-002", "RBI-018"]) - if is_proc_exec: + has_min_severity = False + for cid in candidate_ids: + for r in getattr(self, "all_rules", self.rules): + if r.get("id") == cid and r.get("min_severity") is not None: + has_min_severity = True + break + + if is_proc_exec or has_min_severity: lines = content_str.splitlines() line_idx = candidate.get("line", 1) - 1 line_code = lines[line_idx] if 0 <= line_idx < len(lines) else "" diff --git a/anchor/runtime/relay_protocol.py b/anchor/runtime/relay_protocol.py index a4dff65..34cde6d 100644 --- a/anchor/runtime/relay_protocol.py +++ b/anchor/runtime/relay_protocol.py @@ -74,13 +74,27 @@ def start(self): def _run_loop(self): asyncio.set_event_loop(self.loop) - self.loop.run_until_complete(self._connect_and_listen()) + try: + self.loop.run_until_complete(self._connect_and_listen()) + except asyncio.CancelledError: + pass + finally: + try: + pending = asyncio.all_tasks(self.loop) + if pending: + for task in pending: + task.cancel() + # Run until all pending tasks are fully cancelled + self.loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) + except Exception: + pass + self.loop.close() def stop(self): self.running = False - if self.loop: + if self.loop and self.loop.is_running(): asyncio.run_coroutine_threadsafe(self._close(), self.loop) - self.thread.join(timeout=2.0) + self.thread.join(timeout=3.0) async def _close(self): if self.ws: @@ -88,7 +102,8 @@ async def _close(self): await self.ws.close() except: pass - self.loop.stop() + for task in asyncio.all_tasks(self.loop): + task.cancel() async def _connect_and_listen(self): while self.running: @@ -147,7 +162,11 @@ async def _connect_and_listen(self): "hub_id": self.hub_id } await ws.send(json.dumps(pong)) + except asyncio.CancelledError: + raise except Exception as e: + if not self.running: + break logger.warning(f"[RELAY] Hub connection lost: {e} — reconnecting in 5s...") self.ws = None await asyncio.sleep(5) @@ -282,7 +301,35 @@ def start(self): def _run_loop(self): asyncio.set_event_loop(self.loop) - self.loop.run_until_complete(self._start_server()) + try: + self.loop.run_until_complete(self._start_server()) + except asyncio.CancelledError: + pass + finally: + try: + pending = asyncio.all_tasks(self.loop) + if pending: + for task in pending: + task.cancel() + self.loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) + except Exception: + pass + self.loop.close() + + async def _close(self): + if self.active_ws: + try: + await self.active_ws.close() + except: + pass + if self.server: + try: + self.server.close() + await self.server.wait_closed() + except: + pass + for task in asyncio.all_tasks(self.loop): + task.cancel() async def _start_server(self): async def handler(websocket): @@ -336,16 +383,17 @@ async def handler(websocket): self.active_ws = None self.server = await websockets.serve(handler, self.host, self.port) - while self.running: - await asyncio.sleep(0.1) - self.server.close() - await self.server.wait_closed() + try: + while self.running: + await asyncio.sleep(0.1) + except asyncio.CancelledError: + raise def stop(self): self.running = False - if self.loop: - self.loop.stop() - self.thread.join(timeout=2.0) + if self.loop and self.loop.is_running(): + asyncio.run_coroutine_threadsafe(self._close(), self.loop) + self.thread.join(timeout=3.0) def request_forensic_pull(self, entry_id: str, timeout: float = 3.0) -> Optional[dict]: if not self.active_ws or not self.loop: diff --git a/tests/unit/test_engine.py b/tests/unit/test_engine.py index 30f4a46..fe557c4 100644 --- a/tests/unit/test_engine.py +++ b/tests/unit/test_engine.py @@ -142,3 +142,30 @@ def test_engine_runtime_integration_check(tmp_path): # No AI libraries are imported, so EU-ART12 should not trigger assert len(results_no_ai["violations"]) == 0 +def test_generic_severity_floor_clamping(): + # Setup a dummy rule (TEST-999) with min_severity: "error" + config = { + "rules": [ + { + "id": "TEST-999", + "name": "Custom Command Execution", + "pattern": "os.system\\(.*\\)", + "severity": "blocker", + "min_severity": "error", + "message": "Dangerous command" + } + ] + } + engine = PolicyEngine(config) + from anchor.adapters.python import PythonAdapter + adapter = PythonAdapter() + + # Static call without AI influence (should downgrade, but not below min_severity = error) + content = b"import os\nos.system('ls')\n" + results = engine.scan_file(content, "test.py", adapter) + + assert len(results["violations"]) == 1 + assert results["violations"][0]["id"] == "TEST-999" + # It would normally downgrade to "warning" if no floor existed, but since min_severity is "error", it must hold at "error" + assert results["violations"][0]["severity"] == "error" +