openlatch-provider 0.2.2

Self-service onboarding CLI + runtime daemon for OpenLatch Editors and Providers
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

"""FastAPI app with one detector route.

Run locally:
    uvicorn my_tool.main:app --port 8081

Behind the runtime daemon (preferred for production):
    1. openlatch-provider register
    2. openlatch-provider listen --no-tls --port 8443
    3. openlatch-provider trigger pre_tool_use --tool=Bash \\
           --input='{"command": "ls"}'
"""

from __future__ import annotations

import os
import re
import time

from fastapi import FastAPI, HTTPException, Request

app = FastAPI(title="my-tool", version="0.1.0")

# Replace with your real detection logic. The starter ships a regex match
# against the AKIA AWS-access-key prefix — swap with whatever your tool
# detects (PII patterns, prompt-injection signatures, dangerous shell
# commands, etc.).
_AWS_KEY_PATTERN = re.compile(r"AKIA[0-9A-Z]{16}")


@app.get("/healthz")
async def healthz() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/event")
async def detect(request: Request) -> dict:
    started = time.monotonic()
    payload = await request.json()
    tool_call = payload.get("tool_call", {}) or {}
    tool_input = tool_call.get("input", {}) or {}
    text = str(tool_input)

    if _AWS_KEY_PATTERN.search(text):
        return {
            "riskScore": 99,
            "severityHint": "critical",
            "verdictHint": "deny",
            "ruleId": "aws.access_key",
            "rationaleSummary": "AWS access key detected in tool input",
            "userFacing": {
                "headline": "AWS access key detected",
                "body": "This call would expose an AWS access key. Replace it with a vault reference before retrying.",
                "evidence": [{"label": "aws_key", "valueRedacted": "AKIA****"}],
            },
            "latencyMs": int((time.monotonic() - started) * 1000),
        }

    return {
        "riskScore": 5,
        "severityHint": "low",
        "verdictHint": "allow",
        "rationaleSummary": "No issue detected",
        "latencyMs": int((time.monotonic() - started) * 1000),
    }


def main() -> None:
    import uvicorn

    port = int(os.environ.get("PORT", "8081"))
    uvicorn.run(app, host="127.0.0.1", port=port)