Establish minimal deterministic SentinelAI production boundary

.github/workflows/ci.yml

@@ -0,0 +1,25 @@
+name: CI
+
+on:
+  push:
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: pip
+
+      - name: Install dependencies
+        run: python -m pip install -r requirements-dev.txt
+
+      - name: Run tests
+        run: python -m pytest -q

pyproject.toml

@@ -0,0 +1,3 @@
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["."]

requirements-dev.txt

@@ -0,0 +1,3 @@
+pytest>=8.4,<9.0
+httpx>=0.28,<1.0
+-r services/api/requirements.txt

services/api/app/main.py

@@ -0,0 +1,52 @@
+from fastapi import FastAPI
+
+from services.api.app.models import CheckTransactionRequest, CheckTransactionResponse
+from services.intelligence.sentinel_risk.engine import (
+    Confidence,
+    Reputation,
+    ReputationStatus,
+    TransactionFacts,
+    evaluate_transaction,
+)
+
+POLICY_VERSION = "day13-v3"
+
+app = FastAPI(title="SentinelAI API", version="0.1.0")
+
+
+@app.get("/health", tags=["system"])
+def health() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.post("/v1/check-tx", response_model=CheckTransactionResponse, tags=["risk"])
+def check_transaction(request: CheckTransactionRequest) -> CheckTransactionResponse:
+    tx = TransactionFacts(
+        chain_id=request.chain_id,
+        destination=request.destination,
+        is_unlimited_approval=request.is_unlimited_approval,
+        spender=request.spender,
+    )
+    destination_reputation = Reputation(
+        status=ReputationStatus(request.destination_reputation.status),
+        confidence=Confidence(request.destination_reputation.confidence),
+    )
+    spender_reputation = (
+        Reputation(
+            status=ReputationStatus(request.spender_reputation.status),
+            confidence=Confidence(request.spender_reputation.confidence),
+        )
+        if request.spender_reputation
+        else None
+    )
+
+    decision = evaluate_transaction(
+        tx,
+        destination_reputation,
+        spender_reputation,
+    )
+
+    return CheckTransactionResponse(
+        decision=decision.value,
+        policy_version=POLICY_VERSION,
+    )

services/api/app/models.py

@@ -0,0 +1,38 @@
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from services.api.app.validation import normalize_address
+
+
+class ReputationInput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    status: str = Field(pattern="^(trusted|unknown|malicious)$")
+    confidence: str = Field(pattern="^(low|medium|high)$")
+
+
+class CheckTransactionRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    chain_id: int = Field(gt=0)
+    destination: str
+    is_unlimited_approval: bool = False
+    spender: str | None = None
+    destination_reputation: ReputationInput
+    spender_reputation: ReputationInput | None = None
+
+    @field_validator("destination")
+    @classmethod
+    def validate_destination(cls, value: str) -> str:
+        return normalize_address(value)
+
+    @field_validator("spender")
+    @classmethod
+    def validate_spender(cls, value: str | None) -> str | None:
+        return normalize_address(value) if value is not None else None
+
+
+class CheckTransactionResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    decision: str = Field(pattern="^(allow|warn|block)$")
+    policy_version: str

services/api/app/validation.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+import re
+
+ADDRESS_RE = re.compile(r"^0x[0-9a-fA-F]{40}$")
+
+
+def normalize_address(value: str) -> str:
+    if not ADDRESS_RE.fullmatch(value):
+        raise ValueError("invalid Ethereum address")
+    return value.lower()

services/api/requirements.txt

@@ -0,0 +1,2 @@
+fastapi>=0.116,<1.0
+uvicorn[standard]>=0.35,<1.0

services/intelligence/sentinel_risk/engine.py

@@ -0,0 +1,71 @@
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class Decision(StrEnum):
+    ALLOW = "allow"
+    WARN = "warn"
+    BLOCK = "block"
+
+
+class ReputationStatus(StrEnum):
+    TRUSTED = "trusted"
+    UNKNOWN = "unknown"
+    MALICIOUS = "malicious"
+
+
+class Confidence(StrEnum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+
+
+@dataclass(frozen=True)
+class Reputation:
+    status: ReputationStatus
+    confidence: Confidence
+
+
+@dataclass(frozen=True)
+class TransactionFacts:
+    chain_id: int
+    destination: str
+    is_unlimited_approval: bool = False
+    spender: str | None = None
+
+
+def evaluate_transaction(
+    tx: TransactionFacts,
+    destination_reputation: Reputation,
+    spender_reputation: Reputation | None = None,
+) -> Decision:
+    """Apply the minimal deterministic policy.
+
+    Only high-confidence malicious evidence can BLOCK. Unknown or
+    unavailable evidence never becomes an implicit ALLOW.
+    """
+    if (
+        destination_reputation.status is ReputationStatus.MALICIOUS
+        and destination_reputation.confidence is Confidence.HIGH
+    ):
+        return Decision.BLOCK
+
+    if (
+        tx.is_unlimited_approval
+        and tx.spender is not None
+        and spender_reputation is not None
+        and spender_reputation.status is ReputationStatus.MALICIOUS
+        and spender_reputation.confidence is Confidence.HIGH
+    ):
+        return Decision.BLOCK
+
+    if (
+        destination_reputation.status is ReputationStatus.UNKNOWN
+        or destination_reputation.confidence is Confidence.LOW
+    ):
+        return Decision.WARN
+
+    if tx.is_unlimited_approval:
+        return Decision.WARN
+
+    return Decision.ALLOW

services/telemetry/schema.sql

@@ -0,0 +1,32 @@
+CREATE TABLE IF NOT EXISTS reputation_entries (
+    chain_id BIGINT NOT NULL,
+    address TEXT NOT NULL,
+    status TEXT NOT NULL CHECK (status IN ('trusted', 'unknown', 'malicious')),
+    confidence TEXT NOT NULL CHECK (confidence IN ('low', 'medium', 'high')),
+    source TEXT NOT NULL,
+    observed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (chain_id, address)
+);
+
+CREATE TABLE IF NOT EXISTS decision_events (
+    id UUID PRIMARY KEY,
+    request_id UUID NOT NULL,
+    chain_id BIGINT NOT NULL,
+    from_address TEXT NOT NULL,
+    to_address TEXT NOT NULL,
+    decision TEXT NOT NULL CHECK (decision IN ('allow', 'warn', 'block')),
+    risk_score INTEGER NOT NULL CHECK (risk_score BETWEEN 0 AND 100),
+    category TEXT NOT NULL,
+    confidence TEXT NOT NULL CHECK (confidence IN ('low', 'medium', 'high')),
+    reasons JSONB NOT NULL,
+    latency_ms INTEGER NOT NULL CHECK (latency_ms >= 0),
+    runtime_mode TEXT NOT NULL CHECK (runtime_mode IN ('shadow', 'enforce')),
+    policy_version TEXT NOT NULL,
+    user_action TEXT CHECK (user_action IS NULL OR user_action IN ('accepted', 'cancelled', 'overridden')),
+    wallet_outcome TEXT CHECK (wallet_outcome IS NULL OR wallet_outcome IN ('blocked', 'submitted', 'rejected', 'failed')),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS decision_events_created_idx ON decision_events (created_at);
+CREATE INDEX IF NOT EXISTS decision_events_decision_idx ON decision_events (decision);
+CREATE INDEX IF NOT EXISTS decision_events_request_idx ON decision_events (request_id);

tests/test_api.py

@@ -0,0 +1,78 @@
+from fastapi.testclient import TestClient
+
+from services.api.app.main import app
+
+
+client = TestClient(app)
+VALID_DESTINATION = "0x000000000000000000000000000000000000dEaD"
+VALID_UNKNOWN = "0x0000000000000000000000000000000000000001"
+
+
+def test_high_confidence_malicious_destination_blocks() -> None:
+    response = client.post(
+        "/v1/check-tx",
+        json={
+            "chain_id": 1,
+            "destination": VALID_DESTINATION,
+            "destination_reputation": {
+                "status": "malicious",
+                "confidence": "high",
+            },
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "decision": "block",
+        "policy_version": "day13-v3",
+    }
+
+
+def test_unknown_destination_never_becomes_allow() -> None:
+    response = client.post(
+        "/v1/check-tx",
+        json={
+            "chain_id": 1,
+            "destination": VALID_UNKNOWN,
+            "destination_reputation": {
+                "status": "unknown",
+                "confidence": "low",
+            },
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json()["decision"] == "warn"
+
+
+def test_invalid_ethereum_address_is_rejected() -> None:
+    response = client.post(
+        "/v1/check-tx",
+        json={
+            "chain_id": 1,
+            "destination": "0xdead",
+            "destination_reputation": {
+                "status": "trusted",
+                "confidence": "high",
+            },
+        },
+    )
+
+    assert response.status_code == 422
+
+
+def test_extra_request_fields_are_rejected() -> None:
+    response = client.post(
+        "/v1/check-tx",
+        json={
+            "chain_id": 1,
+            "destination": VALID_DESTINATION,
+            "unexpected": True,
+            "destination_reputation": {
+                "status": "trusted",
+                "confidence": "high",
+            },
+        },
+    )
+
+    assert response.status_code == 422

tests/test_health.py

@@ -1,4 +1,13 @@
-def test_placeholder_health():
-    # Placeholder test so CI has at least one deterministic test target.
-    # Replace with real API health test when app entrypoint is confirmed.
-    assert True
+from fastapi.testclient import TestClient
+
+from services.api.app.main import app
+
+
+client = TestClient(app)
+
+
+def test_health_endpoint() -> None:
+    response = client.get("/health")
+
+    assert response.status_code == 200
+    assert response.json() == {"status": "ok"}