diff --git a/.lint_baselines/falsey_clobber.json b/.lint_baselines/falsey_clobber.json index 45ab20a..c710deb 100644 --- a/.lint_baselines/falsey_clobber.json +++ b/.lint_baselines/falsey_clobber.json @@ -41,20 +41,6 @@ "axonflow/client.py:845:20", "axonflow/client.py:931:20", "axonflow/execution.py:205:19", - "axonflow/interceptors/anthropic.py:134:43", - "axonflow/interceptors/anthropic.py:161:43", - "axonflow/interceptors/bedrock.py:175:43", - "axonflow/interceptors/bedrock.py:206:43", - "axonflow/interceptors/gemini.py:141:39", - "axonflow/interceptors/gemini.py:167:43", - "axonflow/interceptors/gemini.py:228:39", - "axonflow/interceptors/gemini.py:252:43", - "axonflow/interceptors/ollama.py:121:43", - "axonflow/interceptors/ollama.py:150:43", - "axonflow/interceptors/ollama.py:202:43", - "axonflow/interceptors/ollama.py:228:43", - "axonflow/interceptors/openai.py:110:43", - "axonflow/interceptors/openai.py:137:43", "axonflow/masfeat.py:296:23", "axonflow/masfeat.py:297:24", "axonflow/masfeat.py:298:25", diff --git a/CHANGELOG.md b/CHANGELOG.md index 60e65e4..d930cae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,48 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 and tag v{X.Y.Z}. The release workflow's preflight checks the section header matches the tag. --> +## [8.5.1] - 2026-07-09 — Interceptor sync bridge + async-client detection + example fixes + +Hostile-testing sweep ahead of the BukuWarung integration +(getaxonflow/axonflow-enterprise#2861). + +### Fixed + +- **Interceptor sync wrap paths no longer crash inside a running event + loop.** All five provider interceptors (openai, anthropic, bedrock, + gemini, ollama) ran the async governance check with + `loop.run_until_complete`, which raises `RuntimeError: This event loop is + already running` whenever the caller sits inside a running loop (FastAPI + handler, Jupyter, an async app driving a sync provider client) — the + async-adapter-bypass class: the governance check crashed instead of + completing. The shared `run_coroutine_sync` bridge now executes + governance on one persistent background loop (daemon thread) and blocks + the call site until the verdict is in; repeated calls reuse the same + loop, so the HTTP pool stays valid (a throwaway-loop bridge broke the + second call with "Event loop is closed"). +- **`AsyncOpenAI`/`AsyncAnthropic` clients are detected as async again.** + openai>=1 and anthropic decorate their async `create` methods with + plain-`def` wrappers (`@required_args`), so `asyncio.iscoroutinefunction` + returned False and async clients were silently given the SYNC wrap path. + `is_async_callable` follows `__wrapped__` (`inspect.unwrap`) to the real + `async def`. +- **Examples pass `AXONFLOW_USER_TOKEN` and fail honestly.** Enterprise + stacks (`DEPLOYMENT_MODE=enterprise`) validate user tokens as JWTs; + `quickstart`, `gateway_mode` and `openai_integration` passed hardcoded + non-JWT literals and 401'd. All three now read `AXONFLOW_USER_TOKEN`. + `gateway_mode`'s blocked-request demo uses a stacked-SQLi query (blocked + on every stack posture — PII policies default to redact, not block) and + exits non-zero if unexpectedly approved; `openai_integration` prints the + outcome of its policy-block probe instead of ending silently. + +### Added + +- `runtime-e2e/interceptor_sync_bridge/` — live-agent assertion driving a + REAL `openai.OpenAI` client through `wrap_openai_client` from inside a + running loop and from plain sync code (blocked verdict enforced both + ways, no RuntimeError), plus async-detection assertions for + `AsyncOpenAI`. + ## [8.5.0] - 2026-06-09 — Decision Mode PEP: decide → fulfill → forward Adds the SDK analog of the platform PEP client (`platform/shared/pep`, diff --git a/axonflow/_version.py b/axonflow/_version.py index 72f0df3..4ea87e2 100644 --- a/axonflow/_version.py +++ b/axonflow/_version.py @@ -1,3 +1,3 @@ """Single source of truth for the AxonFlow SDK version.""" -__version__ = "8.5.0" +__version__ = "8.5.1" diff --git a/axonflow/interceptors/_sync_bridge.py b/axonflow/interceptors/_sync_bridge.py new file mode 100644 index 0000000..a7fee24 --- /dev/null +++ b/axonflow/interceptors/_sync_bridge.py @@ -0,0 +1,90 @@ +"""Shared sync-over-async bridge for interceptor governance calls. + +Every provider interceptor has a sync wrap path that must run the async +governance check (``AxonFlow.proxy_llm_call``) to completion BEFORE the +provider call. The old per-interceptor pattern:: + + loop = asyncio.get_event_loop() + response = loop.run_until_complete(...) + +raises ``RuntimeError: This event loop is already running`` whenever the +caller sits inside a running loop (FastAPI handler, Jupyter, an async app +driving a sync provider client). A governance check that crashes instead of +completing is the async-adapter-bypass failure class: callers who catch and +continue would proceed ungoverned. + +``run_coroutine_sync`` executes the coroutine on ONE persistent background +event loop (daemon thread, created lazily) and blocks the caller until the +verdict is in — whether or not the calling thread has a running loop. A +persistent loop matters: the AxonFlow client's underlying HTTP pool is +loop-affine, so running each governance call on a throwaway loop +(``asyncio.run`` per call) breaks the SECOND call with "Event loop is +closed". Either way the governance result — allow, block, or error — is +returned/raised synchronously and can never be skipped. + +Sharing note: an ``AxonFlow`` instance's HTTP pool binds to the first loop +that uses it. An instance driven through this bridge (sync-wrapped provider +clients) must not ALSO be awaited on a caller's own loop — construct one +instance per context (one for sync-wrapped clients, one for async-wrapped). +""" + +from __future__ import annotations + +import asyncio +import inspect +import threading +from collections.abc import Coroutine +from typing import Any, TypeVar + +T = TypeVar("T") + + +class _BridgeState: + """Holder for the shared background loop (avoids module-global rebinding).""" + + lock = threading.Lock() + loop: asyncio.AbstractEventLoop | None = None + + +def is_async_callable(fn: Any) -> bool: + """Detect coroutine functions THROUGH ``functools.wraps`` decorators. + + Modern provider SDKs (openai>=1, anthropic) decorate their async + ``create`` methods with plain-``def`` wrappers (e.g. ``@required_args``) + that return the coroutine. ``asyncio.iscoroutinefunction`` on the bound + method is then False, so an ``AsyncOpenAI`` client silently got the SYNC + wrap path. ``inspect.unwrap`` follows ``__wrapped__`` to the real + ``async def``. + """ + if inspect.iscoroutinefunction(fn): + return True + try: + return inspect.iscoroutinefunction(inspect.unwrap(fn)) + except ValueError: # __wrapped__ cycle — be conservative + return False + + +def _get_bridge_loop() -> asyncio.AbstractEventLoop: + """Lazily start the shared background loop (daemon thread).""" + with _BridgeState.lock: + if _BridgeState.loop is None or _BridgeState.loop.is_closed(): + loop = asyncio.new_event_loop() + thread = threading.Thread( + target=loop.run_forever, + name="axonflow-interceptor-bridge", + daemon=True, + ) + thread.start() + _BridgeState.loop = loop + return _BridgeState.loop + + +def run_coroutine_sync(coro: Coroutine[Any, Any, T]) -> T: + """Run *coro* to completion from sync code, inside or outside a loop. + + Always executes on the shared background loop so repeated calls reuse + the same loop (and therefore the same HTTP connection pool), and blocks + the calling thread until the result — the verdict cannot be skipped. + """ + loop = _get_bridge_loop() + return asyncio.run_coroutine_threadsafe(coro, loop).result() diff --git a/axonflow/interceptors/anthropic.py b/axonflow/interceptors/anthropic.py index f146d64..5cdba5d 100644 --- a/axonflow/interceptors/anthropic.py +++ b/axonflow/interceptors/anthropic.py @@ -24,12 +24,12 @@ from __future__ import annotations -import asyncio from collections.abc import Callable from functools import wraps from typing import TYPE_CHECKING, Any, TypeVar from axonflow.exceptions import PolicyViolationError +from axonflow.interceptors._sync_bridge import is_async_callable, run_coroutine_sync from axonflow.interceptors.base import BaseInterceptor if TYPE_CHECKING: @@ -99,16 +99,7 @@ def _extract_prompt(kwargs: dict[str, Any]) -> str: parts.append(block.get("text", "")) return " ".join(parts) - def _get_loop() -> asyncio.AbstractEventLoop: - """Get or create event loop.""" - try: - return asyncio.get_event_loop() - except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - return loop - - if asyncio.iscoroutinefunction(original_create): + if is_async_callable(original_create): @wraps(original_create) async def async_wrapped_create(*args: Any, **kwargs: Any) -> Any: @@ -131,7 +122,11 @@ async def async_wrapped_create(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) # Call original return await original_create(*args, **kwargs) @@ -144,8 +139,7 @@ def sync_wrapped_create(*args: Any, **kwargs: Any) -> Any: prompt = _extract_prompt(kwargs) # Check with AxonFlow (sync) - loop = _get_loop() - response = loop.run_until_complete( + response = run_coroutine_sync( axonflow.proxy_llm_call( user_token=user_token, query=prompt, @@ -158,7 +152,11 @@ def sync_wrapped_create(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) # Call original return original_create(*args, **kwargs) diff --git a/axonflow/interceptors/bedrock.py b/axonflow/interceptors/bedrock.py index 9aa30c2..80ed084 100644 --- a/axonflow/interceptors/bedrock.py +++ b/axonflow/interceptors/bedrock.py @@ -26,13 +26,13 @@ from __future__ import annotations -import asyncio import json from collections.abc import Callable from functools import wraps from typing import TYPE_CHECKING, Any, TypeVar from axonflow.exceptions import PolicyViolationError +from axonflow.interceptors._sync_bridge import run_coroutine_sync from axonflow.interceptors.base import BaseInterceptor if TYPE_CHECKING: @@ -139,15 +139,6 @@ def _extract_prompt(body: Any, _model_id: str) -> str: pass return "" - def _get_loop() -> asyncio.AbstractEventLoop: - """Get or create event loop.""" - try: - return asyncio.get_event_loop() - except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - return loop - # Wrap invoke_model if hasattr(bedrock_client, "invoke_model"): original_invoke = bedrock_client.invoke_model @@ -158,8 +149,7 @@ def sync_wrapped_invoke(*args: Any, **kwargs: Any) -> Any: body = kwargs.get("body", "") prompt = _extract_prompt(body, model_id) - loop = _get_loop() - response = loop.run_until_complete( + response = run_coroutine_sync( axonflow.proxy_llm_call( user_token=user_token, query=prompt, @@ -172,7 +162,11 @@ def sync_wrapped_invoke(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) return original_invoke(*args, **kwargs) @@ -188,8 +182,7 @@ def sync_wrapped_stream(*args: Any, **kwargs: Any) -> Any: body = kwargs.get("body", "") prompt = _extract_prompt(body, model_id) - loop = _get_loop() - response = loop.run_until_complete( + response = run_coroutine_sync( axonflow.proxy_llm_call( user_token=user_token, query=prompt, @@ -203,7 +196,11 @@ def sync_wrapped_stream(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) return original_stream(*args, **kwargs) diff --git a/axonflow/interceptors/gemini.py b/axonflow/interceptors/gemini.py index 419f0d1..bf06283 100644 --- a/axonflow/interceptors/gemini.py +++ b/axonflow/interceptors/gemini.py @@ -21,12 +21,12 @@ from __future__ import annotations -import asyncio from collections.abc import Callable from functools import wraps from typing import TYPE_CHECKING, Any, TypeVar from axonflow.exceptions import PolicyViolationError +from axonflow.interceptors._sync_bridge import run_coroutine_sync from axonflow.interceptors.base import BaseInterceptor if TYPE_CHECKING: @@ -110,22 +110,12 @@ def _extract_prompt(args: tuple[Any, ...], kwargs: dict[str, Any]) -> str: return contents return "" - def _get_loop() -> asyncio.AbstractEventLoop: - """Get or create event loop.""" - try: - return asyncio.get_event_loop() - except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - return loop - @wraps(original_generate) def sync_wrapped_generate(*args: Any, **kwargs: Any) -> Any: prompt = _extract_prompt(args, kwargs) # Check with AxonFlow (sync) - loop = _get_loop() - response = loop.run_until_complete( + response = run_coroutine_sync( axonflow.proxy_llm_call( user_token=user_token, query=prompt, @@ -138,7 +128,11 @@ def sync_wrapped_generate(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) # Call original return original_generate(*args, **kwargs) @@ -164,7 +158,11 @@ async def async_wrapped_generate(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) # Call original return await original_generate_async(*args, **kwargs) @@ -198,20 +196,11 @@ def _wrap_chat_session( original_send = chat_session.send_message original_send_async = getattr(chat_session, "send_message_async", None) - def _get_loop() -> asyncio.AbstractEventLoop: - try: - return asyncio.get_event_loop() - except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - return loop - @wraps(original_send) def sync_wrapped_send(content: Any, **kwargs: Any) -> Any: prompt = content if isinstance(content, str) else str(content) - loop = _get_loop() - response = loop.run_until_complete( + response = run_coroutine_sync( axonflow.proxy_llm_call( user_token=user_token, query=prompt, @@ -225,7 +214,11 @@ def sync_wrapped_send(content: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) return original_send(content, **kwargs) @@ -249,7 +242,11 @@ async def async_wrapped_send(content: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) return await original_send_async(content, **kwargs) diff --git a/axonflow/interceptors/ollama.py b/axonflow/interceptors/ollama.py index 2082578..1663c5c 100644 --- a/axonflow/interceptors/ollama.py +++ b/axonflow/interceptors/ollama.py @@ -26,12 +26,12 @@ from __future__ import annotations -import asyncio from collections.abc import Callable from functools import wraps from typing import TYPE_CHECKING, Any, TypeVar from axonflow.exceptions import PolicyViolationError +from axonflow.interceptors._sync_bridge import run_coroutine_sync from axonflow.interceptors.base import BaseInterceptor if TYPE_CHECKING: @@ -86,15 +86,6 @@ def _extract_chat_prompt(kwargs: dict[str, Any]) -> str: messages = kwargs.get("messages", []) return " ".join(m.get("content", "") for m in messages if isinstance(m, dict)) - def _get_loop() -> asyncio.AbstractEventLoop: - """Get or create event loop.""" - try: - return asyncio.get_event_loop() - except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - return loop - # Wrap chat method if hasattr(ollama_client, "chat"): original_chat = ollama_client.chat @@ -104,8 +95,7 @@ def sync_wrapped_chat(*args: Any, **kwargs: Any) -> Any: prompt = _extract_chat_prompt(kwargs) model = kwargs.get("model", "llama2") - loop = _get_loop() - response = loop.run_until_complete( + response = run_coroutine_sync( axonflow.proxy_llm_call( user_token=user_token, query=prompt, @@ -118,7 +108,11 @@ def sync_wrapped_chat(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) return original_chat(*args, **kwargs) @@ -133,8 +127,7 @@ def sync_wrapped_generate(*args: Any, **kwargs: Any) -> Any: prompt = kwargs.get("prompt", "") model = kwargs.get("model", "llama2") - loop = _get_loop() - response = loop.run_until_complete( + response = run_coroutine_sync( axonflow.proxy_llm_call( user_token=user_token, query=prompt, @@ -147,7 +140,11 @@ def sync_wrapped_generate(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) return original_generate(*args, **kwargs) @@ -199,7 +196,11 @@ async def async_wrapped_chat(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) return await original_chat(*args, **kwargs) @@ -225,7 +226,11 @@ async def async_wrapped_generate(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) return await original_generate(*args, **kwargs) diff --git a/axonflow/interceptors/openai.py b/axonflow/interceptors/openai.py index da69a2f..834bc2f 100644 --- a/axonflow/interceptors/openai.py +++ b/axonflow/interceptors/openai.py @@ -23,12 +23,12 @@ from __future__ import annotations -import asyncio from collections.abc import Callable from functools import wraps from typing import TYPE_CHECKING, Any, TypeVar from axonflow.exceptions import PolicyViolationError +from axonflow.interceptors._sync_bridge import is_async_callable, run_coroutine_sync from axonflow.interceptors.base import BaseInterceptor if TYPE_CHECKING: @@ -77,16 +77,7 @@ def _extract_prompt(kwargs: dict[str, Any]) -> str: messages = kwargs.get("messages", []) return " ".join(m.get("content", "") for m in messages if isinstance(m, dict)) - def _get_loop() -> asyncio.AbstractEventLoop: - """Get or create event loop.""" - try: - return asyncio.get_event_loop() - except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - return loop - - if asyncio.iscoroutinefunction(original_create): + if is_async_callable(original_create): @wraps(original_create) async def async_wrapped_create(*args: Any, **kwargs: Any) -> Any: @@ -107,7 +98,11 @@ async def async_wrapped_create(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) # Call original return await original_create(*args, **kwargs) @@ -120,8 +115,7 @@ def sync_wrapped_create(*args: Any, **kwargs: Any) -> Any: prompt = _extract_prompt(kwargs) # Check with AxonFlow (sync) - loop = _get_loop() - response = loop.run_until_complete( + response = run_coroutine_sync( axonflow.proxy_llm_call( user_token=user_token, query=prompt, @@ -134,7 +128,11 @@ def sync_wrapped_create(*args: Any, **kwargs: Any) -> Any: ) if response.blocked: - raise PolicyViolationError(response.block_reason or "Request blocked by policy") + raise PolicyViolationError( + response.block_reason + if response.block_reason is not None + else "Request blocked by policy" + ) # Call original return original_create(*args, **kwargs) diff --git a/examples/gateway_mode.py b/examples/gateway_mode.py index b68a398..e07ca48 100644 --- a/examples/gateway_mode.py +++ b/examples/gateway_mode.py @@ -13,6 +13,7 @@ import asyncio import os +import sys import time from axonflow import AxonFlow, TokenUsage @@ -51,7 +52,7 @@ async def main() -> None: print("-" * 40) ctx = await axonflow.get_policy_approved_context( - user_token="user-jwt-token", # Your user's JWT + user_token=os.environ.get("AXONFLOW_USER_TOKEN", "user-jwt-token"), # Your user's JWT query="Find patients with recent lab results", data_sources=["postgres"], # MCP connectors to fetch data from context={"department": "cardiology"}, # Additional context @@ -151,17 +152,21 @@ async def blocked_example() -> None: ) as axonflow: print("\n=== Blocked Request Example ===\n") + # A stacked-DROP SQL injection is blocked by the sys_sqli_* static + # policies on every stack posture (PII policies default to redact, + # not block, so a "sensitive-sounding" English query would pass). ctx = await axonflow.get_policy_approved_context( - user_token="user-jwt-token", - query="Show me all social security numbers", # Sensitive query + user_token=os.environ.get("AXONFLOW_USER_TOKEN", "user-jwt-token"), + query="SELECT * FROM users WHERE 1=1; DROP TABLE users;--", ) if not ctx.approved: - print(f"❌ Request blocked!") + print("❌ Request blocked!") print(f" Reason: {ctx.block_reason}") print(f" Policies: {ctx.policies}") else: - print("✅ Request approved (unexpected)") + print("Request approved — expected a policy block for stacked SQLi") + sys.exit(1) if __name__ == "__main__": diff --git a/examples/openai_integration.py b/examples/openai_integration.py index fc1258d..a41c33c 100644 --- a/examples/openai_integration.py +++ b/examples/openai_integration.py @@ -40,7 +40,7 @@ async def main() -> None: wrapped_openai = wrap_openai_client( openai_client, axonflow, - user_token="user-123", # Your user's token + user_token=os.environ.get("AXONFLOW_USER_TOKEN", "user-123"), ) print("OpenAI client wrapped with AxonFlow governance\n") @@ -80,6 +80,10 @@ async def main() -> None: # path). OpenAI errors = no key / rate limit (acceptable noise). # Anything else (e.g. AxonFlow regression) bubbles up. print(f"Request handled: {type(e).__name__}: {e}") + else: + # Whether this query blocks depends on the stack's policy + # posture — say so instead of ending silently. + print("Not blocked by this stack's policies; response received.") if __name__ == "__main__": diff --git a/examples/quickstart.py b/examples/quickstart.py index 611a044..fec1b37 100644 --- a/examples/quickstart.py +++ b/examples/quickstart.py @@ -30,7 +30,7 @@ async def main() -> None: # Execute a simple query with governance print("\n--- Executing governed query ---") response = await client.proxy_llm_call( - user_token="demo-user", + user_token=os.environ.get("AXONFLOW_USER_TOKEN", "demo-user"), query="What is the capital of France?", request_type="chat", ) @@ -55,7 +55,7 @@ def sync_example() -> None: client_secret=os.environ.get("AXONFLOW_CLIENT_SECRET", "demo-secret"), ) as client: result = client.proxy_llm_call( - user_token="demo-user", + user_token=os.environ.get("AXONFLOW_USER_TOKEN", "demo-user"), query="Hello, world!", request_type="chat", ) diff --git a/pyproject.toml b/pyproject.toml index 5dd3ebc..bc076e7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "axonflow" -version = "8.5.0" +version = "8.5.1" description = "AxonFlow Python SDK - Enterprise AI Governance in 3 Lines of Code" readme = "README.md" license = {text = "MIT"} diff --git a/runtime-e2e/interceptor_sync_bridge/test.py b/runtime-e2e/interceptor_sync_bridge/test.py new file mode 100644 index 0000000..7451971 --- /dev/null +++ b/runtime-e2e/interceptor_sync_bridge/test.py @@ -0,0 +1,158 @@ +"""Real-stack assertion: interceptor sync bridge enforces inside a running loop. + +Per CLAUDE.md HARD RULE #0 this test MUST hit a real running AxonFlow agent — +no mocks (the OpenAI client below is the REAL ``openai`` package; it is never +contacted because governance blocks before the provider call). + +Regression background (getaxonflow/axonflow-enterprise#2861): every provider +interceptor's sync wrap path ran the async governance check with +``loop.run_until_complete``, which raises ``RuntimeError: This event loop is +already running`` whenever the caller sits inside a running loop (FastAPI +handler, Jupyter, an async app driving a sync provider client). The +governance check crashed instead of completing — the async-adapter-bypass +class. A second detection bug sent ``AsyncOpenAI`` clients down that sync +path: openai>=1 decorates its async ``create`` with a plain-``def`` wrapper, +so ``asyncio.iscoroutinefunction`` returned False. + +Assertions against a live agent: + + 1. A REAL sync ``openai.OpenAI`` client wrapped with ``wrap_openai_client`` + and driven FROM INSIDE ``asyncio.run`` gets a blocked verdict for a + stacked-SQLi prompt: ``PolicyViolationError`` raised, no RuntimeError, + and the provider is never called (dummy API key would fail loudly). + 2. The same wrapped client called from plain sync code enforces + identically. + 3. A REAL ``openai.AsyncOpenAI`` client is detected as async + (``is_async_callable``) — its wrapped ``create`` is a coroutine + function, so the verdict is awaited on the caller's loop. + +Run locally (after `source /tmp/axonflow-e2e-env.sh` from the enterprise +setup script; requires `pip install openai`): + + AXONFLOW_AGENT_URL=http://localhost:8080 \ + AXONFLOW_CLIENT_ID="$AXONFLOW_CLIENT_ID" \ + AXONFLOW_CLIENT_SECRET="$AXONFLOW_CLIENT_SECRET" \ + AXONFLOW_USER_TOKEN="$AXONFLOW_USER_TOKEN" \ + python3 runtime-e2e/interceptor_sync_bridge/test.py +""" + +from __future__ import annotations + +import asyncio +import inspect +import os +import sys + +import openai + +from axonflow import AxonFlow +from axonflow.exceptions import PolicyViolationError +from axonflow.interceptors._sync_bridge import is_async_callable +from axonflow.interceptors.openai import wrap_openai_client + +AGENT_URL = os.environ.get("AXONFLOW_AGENT_URL", "http://localhost:8080") +CLIENT_ID = os.environ.get("AXONFLOW_CLIENT_ID", "") +CLIENT_SECRET = os.environ.get("AXONFLOW_CLIENT_SECRET", "") +USER_TOKEN = os.environ.get("AXONFLOW_USER_TOKEN", "") + +# Blocked on every stack posture by the sys_sqli_* static policies. +SQLI_PROMPT = "SELECT * FROM users WHERE 1=1; DROP TABLE users;--" + +failures: list[str] = [] + + +def check(name: str, ok: bool, detail: str = "") -> None: + print(f"{'PASS' if ok else 'FAIL'} [{name}] {detail}") + if not ok: + failures.append(name) + + +def make_wrapped_sync_client(axonflow: AxonFlow) -> object: + # REAL openai client, dummy key: if governance ever failed open, the + # provider call would error loudly on auth instead of being blocked. + sync_openai = openai.OpenAI(api_key="sk-dummy-never-called", max_retries=0) + return wrap_openai_client(sync_openai, axonflow, user_token=USER_TOKEN) + + +def drive_blocked(wrapped: object, where: str) -> None: + try: + wrapped.chat.completions.create( # type: ignore[attr-defined] + model="gpt-4", + messages=[{"role": "user", "content": SQLI_PROMPT}], + ) + check(where, False, "call succeeded — governance did not block") + except PolicyViolationError as e: + check(where, True, f"blocked as expected: {e}") + except RuntimeError as e: + check(where, False, f"RuntimeError (bridge regression): {e}") + + +def main() -> None: + if not CLIENT_ID or not CLIENT_SECRET or not USER_TOKEN: + print( + "FAIL: AXONFLOW_CLIENT_ID, AXONFLOW_CLIENT_SECRET and " + "AXONFLOW_USER_TOKEN must be set (source /tmp/axonflow-e2e-env.sh)" + ) + sys.exit(1) + + axonflow = AxonFlow( + endpoint=AGENT_URL, + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + ) + wrapped = make_wrapped_sync_client(axonflow) + + # 1. Sync provider client driven from INSIDE a running event loop. + async def driver() -> None: + drive_blocked(wrapped, "sync-client-inside-running-loop") + + asyncio.run(driver()) + + # 2. Same wrapped client from plain sync code. + drive_blocked(wrapped, "sync-client-plain-sync") + + # 3. AsyncOpenAI must be detected as async through openai's decorators. + async_openai = openai.AsyncOpenAI(api_key="sk-dummy-never-called", max_retries=0) + detected = is_async_callable(async_openai.chat.completions.create) + check( + "async-client-detection", + detected, + "AsyncOpenAI.create detected as coroutine function through @required_args", + ) + if detected: + # Fresh AxonFlow instance for the async leg: httpx connection pools + # are loop-affine, so one instance must not be shared between the + # sync bridge loop and a caller's own loop (see _sync_bridge docs). + axonflow_async = AxonFlow( + endpoint=AGENT_URL, + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + ) + wrapped_async = wrap_openai_client(async_openai, axonflow_async, user_token=USER_TOKEN) + installed = wrapped_async.chat.completions.create + check( + "async-client-gets-async-wrapper", + inspect.iscoroutinefunction(installed), + "wrapped create is async def (verdict awaited on caller's loop)", + ) + + async def async_driver() -> None: + try: + await wrapped_async.chat.completions.create( + model="gpt-4", + messages=[{"role": "user", "content": SQLI_PROMPT}], + ) + check("async-client-blocked", False, "call succeeded — not blocked") + except PolicyViolationError as e: + check("async-client-blocked", True, f"blocked as expected: {e}") + + asyncio.run(async_driver()) + + if failures: + print(f"RESULT: FAIL ({len(failures)}): {failures}") + sys.exit(1) + print("RESULT: PASS (5/5)") + + +if __name__ == "__main__": + main() diff --git a/tests/test_interceptors_execution.py b/tests/test_interceptors_execution.py index bb92880..80eed77 100644 --- a/tests/test_interceptors_execution.py +++ b/tests/test_interceptors_execution.py @@ -1062,3 +1062,164 @@ async def test_ollama_async_without_generate_method(self) -> None: result = await wrapped.chat(model="llama2", messages=[{"content": "Hi"}]) assert result is not None + + +class TestSyncWrapInsideRunningLoop: + """Sync wrap paths must enforce from inside a running event loop. + + Regression for the async-adapter-bypass class: the sync bridges used + ``loop.run_until_complete`` which raises ``RuntimeError: This event loop + is already running`` whenever the caller sits inside a running loop + (FastAPI handler, Jupyter, an async app driving a sync provider client). + The governance check crashed instead of completing — and a caller that + caught the RuntimeError could proceed ungoverned. The shared + ``run_coroutine_sync`` bridge must deliver the verdict either way. + """ + + @staticmethod + def _axonflow(blocked: bool) -> MagicMock: + mock_axonflow = MagicMock() + mock_axonflow.proxy_llm_call = AsyncMock( + return_value=MagicMock(blocked=blocked, block_reason="nope" if blocked else None) + ) + return mock_axonflow + + def test_openai_sync_create_allowed_inside_running_loop(self) -> None: + mock_openai = MagicMock() + mock_openai.chat.completions.create = MagicMock(return_value={"ok": True}) + wrapped = wrap_openai_client(mock_openai, self._axonflow(blocked=False), user_token="u") + + async def driver() -> Any: + # Sync provider client used from async code — the pre-fix bridge + # raised RuntimeError here. + return wrapped.chat.completions.create( + model="gpt-4", messages=[{"role": "user", "content": "Hello"}] + ) + + assert asyncio.run(driver()) == {"ok": True} + + def test_openai_sync_create_blocked_inside_running_loop(self) -> None: + mock_openai = MagicMock() + original_create = MagicMock() + mock_openai.chat.completions.create = original_create + wrapped = wrap_openai_client(mock_openai, self._axonflow(blocked=True)) + + async def driver() -> None: + wrapped.chat.completions.create( + model="gpt-4", messages=[{"role": "user", "content": "Bad"}] + ) + + with pytest.raises(PolicyViolationError): + asyncio.run(driver()) + original_create.assert_not_called() + + def test_anthropic_sync_create_blocked_inside_running_loop(self) -> None: + from axonflow.interceptors.anthropic import wrap_anthropic_client + + mock_anthropic = MagicMock() + original_create = MagicMock() + mock_anthropic.messages.create = original_create + wrapped = wrap_anthropic_client(mock_anthropic, self._axonflow(blocked=True)) + + async def driver() -> None: + wrapped.messages.create(model="claude-3", messages=[{"role": "user", "content": "Bad"}]) + + with pytest.raises(PolicyViolationError): + asyncio.run(driver()) + original_create.assert_not_called() + + def test_bedrock_sync_invoke_blocked_inside_running_loop(self) -> None: + mock_bedrock = MagicMock() + original_invoke = MagicMock() + mock_bedrock.invoke_model = original_invoke + wrapped = wrap_bedrock_client(mock_bedrock, self._axonflow(blocked=True)) + + async def driver() -> None: + wrapped.invoke_model( + modelId="anthropic.claude-3", + body=json.dumps({"messages": [{"role": "user", "content": "Bad"}]}), + ) + + with pytest.raises(PolicyViolationError): + asyncio.run(driver()) + original_invoke.assert_not_called() + + def test_gemini_sync_generate_blocked_inside_running_loop(self) -> None: + mock_model = MagicMock() + original_generate = MagicMock() + mock_model.generate_content = original_generate + wrapped = wrap_gemini_model(mock_model, self._axonflow(blocked=True)) + + async def driver() -> None: + wrapped.generate_content("Bad prompt") + + with pytest.raises(PolicyViolationError): + asyncio.run(driver()) + original_generate.assert_not_called() + + def test_ollama_sync_chat_blocked_inside_running_loop(self) -> None: + mock_ollama = MagicMock() + original_chat = MagicMock() + mock_ollama.chat = original_chat + mock_ollama.generate = MagicMock() + wrapped = wrap_ollama_client(mock_ollama, self._axonflow(blocked=True)) + + async def driver() -> None: + wrapped.chat(model="llama2", messages=[{"role": "user", "content": "Bad"}]) + + with pytest.raises(PolicyViolationError): + asyncio.run(driver()) + original_chat.assert_not_called() + + +class TestAsyncDetectionThroughDecorators: + """AsyncOpenAI-style clients must get the ASYNC wrap path. + + openai>=1 / anthropic decorate their async ``create`` methods with a + plain-``def`` wrapper (``@required_args``) that returns the coroutine — + ``asyncio.iscoroutinefunction`` on the bound method is False, so the + interceptor used to give an async client the sync wrap path. + ``is_async_callable`` follows ``__wrapped__``. + """ + + @staticmethod + def _decorated_async_create() -> Any: + """Mimic openai's @required_args: plain-def wrapper over async def.""" + import functools + + async def create(*args: Any, **kwargs: Any) -> Any: + return {"choices": [{"message": {"content": "hi"}}]} + + @functools.wraps(create) + def wrapper(*args: Any, **kwargs: Any) -> Any: + return create(*args, **kwargs) + + return wrapper + + def test_is_async_callable_sees_through_wraps(self) -> None: + from axonflow.interceptors._sync_bridge import is_async_callable + + assert is_async_callable(self._decorated_async_create()) + + def plain(*args: Any, **kwargs: Any) -> Any: + return {} + + assert not is_async_callable(plain) + + @pytest.mark.asyncio + async def test_openai_wrap_picks_async_path_for_decorated_create(self) -> None: + mock_axonflow = MagicMock() + mock_axonflow.proxy_llm_call = AsyncMock( + return_value=MagicMock(blocked=False, block_reason=None) + ) + mock_openai = MagicMock() + mock_openai.chat.completions.create = self._decorated_async_create() + + wrapped = wrap_openai_client(mock_openai, mock_axonflow, user_token="u") + # The async path installs an `async def` wrapper; awaiting it must work + # end-to-end on the caller's loop. + result = await wrapped.chat.completions.create( + model="gpt-4", messages=[{"role": "user", "content": "Hello"}] + ) + assert result == {"choices": [{"message": {"content": "hi"}}]} + mock_axonflow.proxy_llm_call.assert_awaited()