icebox-gov 0.2.6

Runtime governance for autonomous offensive security — the single seam every human operator, REST client, and LLM agent must pass through before anything touches a target.
Documentation 
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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168

"""Unit tests for the ICEBOX Python SDK (HTTP transport).

Spins up a real in-process HTTP server so no ICEBOX daemon is required.
"""

import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import pytest

import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "python"))

from icebox import IceboxClient, IceboxError
from icebox.tools import openai_tools, dispatch_tool_call


MODULES = [
    {"name": "mysql_scanner", "kind": "scanner", "description": "MySQL scanner"},
    {"name": "vuln_scanner",  "kind": "analysis", "description": "Dep vuln scanner"},
]

MODULE_DETAIL = {
    "name": "mysql_scanner",
    "kind": "scanner",
    "description": "MySQL scanner",
    "author": "ICEBOX",
    "options": {"host": "", "port": 3306, "check_defaults": False},
    "target": None,
    "in_scope": None,
    "charter_accepted": False,
}

APPROVALS = [
    {"id": 1, "module": "mysql_scanner", "target": "10.0.0.1", "reason": "destructive", "status": "Pending"},
]

RUN_RESULT = {"job_id": 42, "success": True, "data": {}, "preflight": None, "error": None}
AUDIT = [{"at": 1000, "target": "10.0.0.1", "module": "mysql_scanner", "decision": "Allow"}]


class _Handler(BaseHTTPRequestHandler):
    def log_message(self, *_):
        pass

    def _respond(self, body, status=200):
        data = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", len(data))
        self.end_headers()
        self.wfile.write(data)

    def do_GET(self):
        routes = {
            "/api/v1/modules": MODULES,
            "/api/v1/modules/mysql_scanner": MODULE_DETAIL,
            "/api/v1/approvals": APPROVALS,
            "/api/v1/audit": AUDIT,
        }
        path = self.path.split("?")[0]
        if path in routes:
            self._respond(routes[path])
        else:
            self._respond({"error": "not found"}, 404)

    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        self.rfile.read(length)
        if self.path == "/api/v1/modules/mysql_scanner/run":
            self._respond(RUN_RESULT)
        elif "/approve" in self.path:
            self._respond("approved")
        elif "/deny" in self.path:
            self._respond("denied")
        else:
            self._respond({"error": "not found"}, 404)


@pytest.fixture(scope="module")
def client():
    server = HTTPServer(("127.0.0.1", 0), _Handler)
    port = server.server_address[1]
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    yield IceboxClient(f"http://127.0.0.1:{port}")
    server.shutdown()


def test_list_modules(client):
    mods = client.list_modules()
    assert len(mods) == 2
    assert mods[0]["name"] == "mysql_scanner"


def test_get_module(client):
    detail = client.get_module("mysql_scanner")
    assert detail["author"] == "ICEBOX"
    assert "options" in detail


def test_run_module(client):
    result = client.run_module("mysql_scanner", target="10.0.0.1", sandbox=True)
    assert result["job_id"] == 42
    assert result["success"] is True


def test_pending_approvals(client):
    pending = client.pending_approvals()
    assert len(pending) == 1
    assert pending[0]["id"] == 1


def test_approve(client):
    resp = client.approve(1)
    assert resp == "approved"


def test_deny(client):
    resp = client.deny(1)
    assert resp == "denied"


def test_audit(client):
    records = client.audit(n=5)
    assert len(records) == 1
    assert records[0]["module"] == "mysql_scanner"


def test_openai_tools_schema(client):
    tools = openai_tools(client)
    assert len(tools) == 2
    mysql = next(t for t in tools if t["function"]["name"] == "icebox_mysql_scanner")
    params = mysql["function"]["parameters"]
    assert params["type"] == "object"
    assert "target" in params["properties"]
    assert "sandbox" in params["properties"]
    assert "target" in params["required"]
    assert mysql["type"] == "function"


def test_openai_tool_schema_types(client):
    tools = openai_tools(client)
    mysql = next(t for t in tools if t["function"]["name"] == "icebox_mysql_scanner")
    props = mysql["function"]["parameters"]["properties"]
    assert props["port"]["type"] == "integer"
    assert props["check_defaults"]["type"] == "boolean"


def test_dispatch_tool_call(client):
    result = dispatch_tool_call(
        client, "icebox_mysql_scanner", {"target": "10.0.0.1", "sandbox": True}
    )
    assert result["job_id"] == 42


def test_dispatch_tool_call_bad_name(client):
    with pytest.raises(ValueError, match="Not an ICEBOX tool"):
        dispatch_tool_call(client, "something_else", {"target": "x"})


def test_governance_compat(client):
    from icebox import Governance
    gov = Governance({"url": client._base})
    assert isinstance(gov.pending(), list)
    assert isinstance(gov.audit_json(), list)