diff --git a/api-gateway/src/main.py b/api-gateway/src/main.py index 066411bf..ffd52135 100644 --- a/api-gateway/src/main.py +++ b/api-gateway/src/main.py @@ -35,6 +35,7 @@ from pydantic import BaseModel from security import verify_public_membrane from starlette.middleware.cors import CORSMiddleware +from starlette.types import ASGIApp, Message, Receive, Scope, Send from telemetry import init_telemetry from config import get_settings @@ -130,6 +131,49 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: allow_headers=["*"], ) + +class BodyCachingMiddleware: + """Pure ASGI middleware that reads and replays the request body. + + Stores raw bytes in scope["body_cache"] so that security dependencies + can read the body even after File(...) params have consumed the stream. + """ + + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": + await self.app(scope, receive, send) + return + + chunks: list[bytes] = [] + more_body = True + while more_body: + message: Message = await receive() + chunks.append(message.get("body", b"")) + more_body = message.get("more_body", False) + + scope["body_cache"] = b"".join(chunks) + + received = False + + async def replayable_receive() -> Message: + nonlocal received + if not received: + received = True + return { + "type": "http.request", + "body": scope["body_cache"], + "more_body": False, + } + return {"type": "http.disconnect"} + + await self.app(scope, replayable_receive, send) + + +app.add_middleware(BodyCachingMiddleware) + FastAPIInstrumentor.instrument_app(app) REQUEST_ID_METADATA_KEY = "x-request-id" @@ -141,11 +185,6 @@ async def request_id_middleware( ) -> Response: request_id = str(uuid.uuid4()) bind_request_id(request_id) - # Pre-cache the raw body so that request.body() in the security dependency - # and request.form() for File(...) parameters both see the same bytes. - # Without this, File(...) consuming the ASGI stream first causes - # request.body() to raise RuntimeError("Stream consumed"). - await request.body() try: return await call_next(request) finally: diff --git a/api-gateway/src/security.py b/api-gateway/src/security.py index 0d68ca93..6396f063 100644 --- a/api-gateway/src/security.py +++ b/api-gateway/src/security.py @@ -25,10 +25,17 @@ async def _parse_request_body(request: Request) -> bytes: """Read body bytes, parse JSON if applicable, and cache in request.state.parsed_body. + Reads from scope["body_cache"] set by BodyCachingMiddleware to avoid + RuntimeError("Stream consumed") when File(...) params consume the stream + before this dependency runs. + Returns raw body bytes for callers that need them (e.g. for signature hashing). Raises HTTPException 400 on malformed JSON. """ - body_bytes = await request.body() + # Use body cached by BodyCachingMiddleware (stored in ASGI scope). + # Falls back to b"" (safe: produces 401 not 500) if middleware not active. + body_bytes: bytes = request.scope.get("body_cache", b"") + content_type = request.headers.get("content-type", "") if body_bytes and "application/json" in content_type: try: @@ -37,6 +44,7 @@ async def _parse_request_body(request: Request) -> bytes: raise HTTPException(status_code=400, detail="Invalid JSON body") from None else: request.state.parsed_body = {} + return body_bytes diff --git a/core/src/hive/membrane/main.py b/core/src/hive/membrane/main.py index ccd101b3..475f6bb6 100644 --- a/core/src/hive/membrane/main.py +++ b/core/src/hive/membrane/main.py @@ -137,7 +137,8 @@ async def inspect_outbound(self, decision: Intent, context: Context) -> Intent: ) return Intent( action=cast(ActionType, ActionType.ACTION_TYPE_REJECT), - reasoning=decision.reasoning + " [MEMBRANE: KYC compliance failure]", + reasoning=decision.reasoning + + " [MEMBRANE: KYC compliance failure]", rwa_vault=rwa_intent, ) return decision @@ -155,7 +156,8 @@ async def inspect_outbound(self, decision: Intent, context: Context) -> Intent: ) return Intent( action=cast(ActionType, ActionType.ACTION_TYPE_REJECT), - reasoning=decision.reasoning + " [MEMBRANE: high-risk trade blocked]", + reasoning=decision.reasoning + + " [MEMBRANE: high-risk trade blocked]", trade=trade_intent, ) return decision diff --git a/core/src/hive/metabolism/errors.py b/core/src/hive/metabolism/errors.py index a84f48c5..888ae171 100644 --- a/core/src/hive/metabolism/errors.py +++ b/core/src/hive/metabolism/errors.py @@ -2,10 +2,14 @@ Metabolic Errors: Custom exceptions for the Hive's ATCG loop. """ + class MetabolicError(Exception): """Base class for all metabolic errors.""" + pass + class MetabolicSecurityError(MetabolicError): """Raised when a security guardrail or membrane enforcement fails.""" + pass diff --git a/core/src/hive/proteins/transaction/skill.py b/core/src/hive/proteins/transaction/skill.py index 45892bff..d61c45a7 100644 --- a/core/src/hive/proteins/transaction/skill.py +++ b/core/src/hive/proteins/transaction/skill.py @@ -227,7 +227,9 @@ async def _execute_rwa_collateral(self, params: dict[str, Any]) -> Observation: # 1. Access context metadata for security enforcement context = params.get("_context") if not context: - raise MetabolicSecurityError("Security context missing: HiveContext required") + raise MetabolicSecurityError( + "Security context missing: HiveContext required" + ) # Extract metadata from Context (google.protobuf.Struct) metadata = ( @@ -237,7 +239,9 @@ async def _execute_rwa_collateral(self, params: dict[str, Any]) -> Observation: aml_risk = metadata.get("aml_risk") # 2. Strict C2C9 Enforcement logic - required_kyc = self.settings.required_kyc_status if self.settings else "APPROVED" + required_kyc = ( + self.settings.required_kyc_status if self.settings else "APPROVED" + ) required_aml = self.settings.required_aml_risk if self.settings else "LOW" if kyc_status != required_kyc or aml_risk != required_aml: diff --git a/core/src/hive/proteins/transaction/solana_engine.py b/core/src/hive/proteins/transaction/solana_engine.py index 3fff7f16..d1f128c5 100644 --- a/core/src/hive/proteins/transaction/solana_engine.py +++ b/core/src/hive/proteins/transaction/solana_engine.py @@ -40,7 +40,9 @@ def __init__(self, private_key_base58: str, rpc_url: str, usdc_mint: str): def _derive_ata(self, owner: Pubkey, mint: Pubkey) -> Pubkey: return get_associated_token_address(owner, mint) - async def execute_rwa_collateral(self, wallet_address: str, amount_usdc: float) -> str: + async def execute_rwa_collateral( + self, wallet_address: str, amount_usdc: float + ) -> str: """ Execute an SPL Token transfer (USDC) from the Hive Treasury to the user's wallet. Simulates the release of a stablecoin loan. @@ -49,14 +51,18 @@ async def execute_rwa_collateral(self, wallet_address: str, amount_usdc: float) user_ata = self._derive_ata(user_pubkey, self.usdc_mint_pubkey) # 1. Verification of Token Decimals (Robustness check) - mint_info = (await self.async_rpc_client.get_token_supply(self.usdc_mint_pubkey)).value + mint_info = ( + await self.async_rpc_client.get_token_supply(self.usdc_mint_pubkey) + ).value if mint_info.decimals != USDC_DECIMALS: logger.error( "usdc_decimal_mismatch", expected=USDC_DECIMALS, actual=mint_info.decimals, ) - raise ValueError(f"USDC decimal mismatch: expected {USDC_DECIMALS}, got {mint_info.decimals}") + raise ValueError( + f"USDC decimal mismatch: expected {USDC_DECIMALS}, got {mint_info.decimals}" + ) amount_raw = int(amount_usdc * (10**USDC_DECIMALS)) @@ -81,7 +87,9 @@ async def execute_rwa_collateral(self, wallet_address: str, amount_usdc: float) ) # Build message and transaction using solders - recent_blockhash = (await self.async_rpc_client.get_latest_blockhash()).value.blockhash + recent_blockhash = ( + await self.async_rpc_client.get_latest_blockhash() + ).value.blockhash msg = Message([transfer_ix], self.keypair.pubkey()) tx = Transaction([self.keypair], msg, recent_blockhash) diff --git a/core/src/hive/transformer/main.py b/core/src/hive/transformer/main.py index f5c3771e..27071a42 100644 --- a/core/src/hive/transformer/main.py +++ b/core/src/hive/transformer/main.py @@ -186,17 +186,31 @@ def _build_rwa_context( """ vision_report: dict[str, Any] = _as_dict(metadata.get("vision_report")) # type: ignore[assignment] wallet_address = str(metadata.get("wallet_address", "")) - kyc_status_str = "true" if metadata.get("kyc_status", "false") == "true" else "false" + kyc_status_str = ( + "true" if metadata.get("kyc_status", "false") == "true" else "false" + ) ltv_ratio_str = str(self._rwa_ltv_ratio()) six_rates: dict[str, Any] = _as_dict(metadata.get("six_rates")) # type: ignore[assignment] system_vitals: dict[str, Any] = _as_dict(metadata.get("vitals")) # type: ignore[assignment] - return vision_report, wallet_address, kyc_status_str, ltv_ratio_str, six_rates, system_vitals + return ( + vision_report, + wallet_address, + kyc_status_str, + ltv_ratio_str, + six_rates, + system_vitals, + ) async def _think_rwa(self, context: Context, metadata: dict[str, Any]) -> Intent: """RWA path: invoke AppraiseAndVerifyRWA via the reasoning protein.""" - vision_report, wallet_address, kyc_status_str, ltv_ratio_str, six_rates, system_vitals = ( - self._build_rwa_context(metadata) - ) + ( + vision_report, + wallet_address, + kyc_status_str, + ltv_ratio_str, + six_rates, + system_vitals, + ) = self._build_rwa_context(metadata) obs = await self.registry.execute( "reasoning", @@ -253,7 +267,9 @@ async def _think_rwa(self, context: Context, metadata: dict[str, Any]) -> Intent is_rejected = compliance_status == "REJECTED" action = ( - ActionType.ACTION_TYPE_REJECT if is_rejected else ActionType.ACTION_TYPE_ACCEPT + ActionType.ACTION_TYPE_REJECT + if is_rejected + else ActionType.ACTION_TYPE_ACCEPT ) if is_rejected: @@ -354,9 +370,14 @@ async def _think_trade(self, context: Context, metadata: dict[str, Any]) -> Inte ) risk_threshold = self._trade_risk_threshold() - is_high_risk = risk_score > risk_threshold or "REJECTED_HIGH_RISK" in trade_intent.reasoning + is_high_risk = ( + risk_score > risk_threshold + or "REJECTED_HIGH_RISK" in trade_intent.reasoning + ) action = ( - ActionType.ACTION_TYPE_REJECT if is_high_risk else ActionType.ACTION_TYPE_ACCEPT + ActionType.ACTION_TYPE_REJECT + if is_high_risk + else ActionType.ACTION_TYPE_ACCEPT ) if is_high_risk: diff --git a/core/src/hive/transformer/signatures.py b/core/src/hive/transformer/signatures.py index c250e914..63e1f965 100644 --- a/core/src/hive/transformer/signatures.py +++ b/core/src/hive/transformer/signatures.py @@ -66,7 +66,7 @@ class AppraiseAndVerifyRWA(dspy.Signature): vision_report: str = dspy.InputField( desc="JSON string from Gemma 3 describing the physical asset: type, condition, " - "weight/size, estimated value, and any anomalies detected." + "weight/size, estimated value, and any anomalies detected." ) wallet_address: str = dspy.InputField( desc="Solana wallet address of the counterparty submitting the asset." @@ -76,11 +76,11 @@ class AppraiseAndVerifyRWA(dspy.Signature): ) six_rates: str = dspy.InputField( desc="JSON string of SIX data partner FX and precious metals rates " - "(e.g. XAU/USD, XAG/USD, EUR/USD) used for asset valuation." + "(e.g. XAU/USD, XAG/USD, EUR/USD) used for asset valuation." ) ltv_ratio: str = dspy.InputField( desc="Loan-to-value ratio as a plain float string (e.g. '0.60'). " - "collateral_value_usd = appraised_value_usd * ltv_ratio." + "collateral_value_usd = appraised_value_usd * ltv_ratio." ) system_vitals: str = dspy.InputField( desc="JSON string with system health metrics (CPU, memory, latency)." @@ -94,7 +94,7 @@ class AppraiseAndVerifyRWA(dspy.Signature): ) violation_code: str = dspy.OutputField( desc='Empty string if approved. One of: "KYC_MISSING", "AML_SUSPICIOUS", ' - '"ASSET_UNVERIFIED" if rejected.' + '"ASSET_UNVERIFIED" if rejected.' ) appraised_value_usd: str = dspy.OutputField( desc='Appraised asset value in USD as a plain float string. "0.0" if rejected.' @@ -104,9 +104,9 @@ class AppraiseAndVerifyRWA(dspy.Signature): ) vault_intent_json: str = dspy.OutputField( desc="A single valid JSON object with fields: vault_id, asset_identifier, " - "asset_domain, appraised_value_usd (float), ltv_ratio (float), " - "collateral_value_usd (float), stablecoin_currency, wallet_address, reasoning. " - "No markdown fences, no extra text." + "asset_domain, appraised_value_usd (float), ltv_ratio (float), " + "collateral_value_usd (float), stablecoin_currency, wallet_address, reasoning. " + "No markdown fences, no extra text." ) @@ -141,7 +141,7 @@ class GenerateTradeRisk(dspy.Signature): ) risk_threshold: str = dspy.InputField( desc="Maximum acceptable risk score as a plain float string (e.g. '0.10'). " - "Trades with risk_score above this value must be rejected." + "Trades with risk_score above this value must be rejected." ) think: str = dspy.OutputField( @@ -150,9 +150,7 @@ class GenerateTradeRisk(dspy.Signature): risk_score: str = dspy.OutputField( desc="Float 0.0–1.0 as a plain string (e.g. '0.07'). Lower is safer." ) - risk_category: str = dspy.OutputField( - desc="One of: LOW, MEDIUM, HIGH, CRITICAL." - ) + risk_category: str = dspy.OutputField(desc="One of: LOW, MEDIUM, HIGH, CRITICAL.") class GenerateTradeIntent(dspy.Signature): @@ -209,7 +207,7 @@ class GenerateTradeIntent(dspy.Signature): ) risk_threshold: str = dspy.InputField( desc="Maximum acceptable risk score as a plain float string (e.g. '0.10'). " - "Trades with risk_score above this value must be rejected." + "Trades with risk_score above this value must be rejected." ) trade_intent_json: str = dspy.OutputField( diff --git a/core/tests/test_rwa_transformer.py b/core/tests/test_rwa_transformer.py index ea717152..a79eb581 100644 --- a/core/tests/test_rwa_transformer.py +++ b/core/tests/test_rwa_transformer.py @@ -149,7 +149,11 @@ def test_rwa_negotiator_approved(mocker): mocker.patch.object(negotiator.appraise, "forward", return_value=mock_pred) result = negotiator.forward( - vision_report={"asset_type": "GOLD", "weight_oz": 2.0, "condition": "excellent"}, + vision_report={ + "asset_type": "GOLD", + "weight_oz": 2.0, + "condition": "excellent", + }, wallet_address="abc12345xyz", kyc_status="true", six_rates={"XAU_USD": 31000.0}, @@ -262,9 +266,9 @@ async def test_transformer_rwa_takes_priority_over_trade(): intent = await transformer.think(context) params_name, _ = betterproto.which_one_of(intent, "params") - assert params_name == "rwa_vault", ( - f"RWA path should take priority over trade path, got '{params_name}'" - ) + assert ( + params_name == "rwa_vault" + ), f"RWA path should take priority over trade path, got '{params_name}'" # Verify reasoning protein was called with 'rwa', not 'trade' # skill.execute(intent, params) — intent is the first positional arg call_args = mock_reasoning.execute.call_args diff --git a/uv.lock b/uv.lock index f0993f29..89fd471a 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = "==3.12.*" resolution-markers = [ "sys_platform == 'win32'", @@ -472,7 +472,7 @@ dependencies = [ [package.metadata] requires-dist = [ { name = "aura-core", editable = "packages/aura-core" }, - { name = "core", virtual = "core" }, + { name = "core", editable = "core" }, { name = "fastmcp", specifier = "==2.12.3" }, { name = "httpx" }, { name = "nats-py", specifier = ">=2.9.0" }, @@ -820,7 +820,7 @@ wheels = [ [[package]] name = "core" version = "0.1.0" -source = { virtual = "core" } +source = { editable = "core" } dependencies = [ { name = "alembic" }, { name = "aura-core" }, @@ -1483,7 +1483,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f9/c8/9d76a66421d1ae24340dfae7e79c313957f6e3195c144d2c73333b5bfe34/greenlet-3.3.1-cp312-cp312-macosx_11_0_universal2.whl", hash = "sha256:7e806ca53acf6d15a888405880766ec84721aa4181261cd11a457dfe9a7a4975", size = 276443, upload-time = "2026-01-23T15:30:10.066Z" }, { url = "https://files.pythonhosted.org/packages/81/99/401ff34bb3c032d1f10477d199724f5e5f6fbfb59816ad1455c79c1eb8e7/greenlet-3.3.1-cp312-cp312-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:d842c94b9155f1c9b3058036c24ffb8ff78b428414a19792b2380be9cecf4f36", size = 597359, upload-time = "2026-01-23T16:00:57.394Z" }, { url = "https://files.pythonhosted.org/packages/2b/bc/4dcc0871ed557792d304f50be0f7487a14e017952ec689effe2180a6ff35/greenlet-3.3.1-cp312-cp312-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:20fedaadd422fa02695f82093f9a98bad3dab5fcda793c658b945fcde2ab27ba", size = 607805, upload-time = "2026-01-23T16:05:28.068Z" }, - { url = "https://files.pythonhosted.org/packages/3b/cd/7a7ca57588dac3389e97f7c9521cb6641fd8b6602faf1eaa4188384757df/greenlet-3.3.1-cp312-cp312-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:c620051669fd04ac6b60ebc70478210119c56e2d5d5df848baec4312e260e4ca", size = 622363, upload-time = "2026-01-23T16:15:54.754Z" }, { url = "https://files.pythonhosted.org/packages/cf/05/821587cf19e2ce1f2b24945d890b164401e5085f9d09cbd969b0c193cd20/greenlet-3.3.1-cp312-cp312-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:14194f5f4305800ff329cbf02c5fcc88f01886cadd29941b807668a45f0d2336", size = 609947, upload-time = "2026-01-23T15:32:51.004Z" }, { url = "https://files.pythonhosted.org/packages/a4/52/ee8c46ed9f8babaa93a19e577f26e3d28a519feac6350ed6f25f1afee7e9/greenlet-3.3.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:7b2fe4150a0cf59f847a67db8c155ac36aed89080a6a639e9f16df5d6c6096f1", size = 1567487, upload-time = "2026-01-23T16:04:22.125Z" }, { url = "https://files.pythonhosted.org/packages/8f/7c/456a74f07029597626f3a6db71b273a3632aecb9afafeeca452cfa633197/greenlet-3.3.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:49f4ad195d45f4a66a0eb9c1ba4832bb380570d361912fa3554746830d332149", size = 1636087, upload-time = "2026-01-23T15:33:47.486Z" },