diff --git a/hub/client/permission_hook.py b/hub/client/permission_hook.py index 7f983c1..78e6462 100755 --- a/hub/client/permission_hook.py +++ b/hub/client/permission_hook.py @@ -13,8 +13,9 @@ Returns: stdout: the PermissionRequest decision JSON: allow only when a pinned passkey signed exactly this tool, command, cwd, session and ask id; deny when the operator denied in the kvb-hub. Nothing on timeout, - error, a missing or invalid signature or a hub that is down, so - Claude Code shows its own dialog. The exit code is 0. + error, a missing or invalid signature, a value longer than the hub + takes or a hub that is down, so Claude Code shows its own dialog. + The exit code is 0. """ from __future__ import annotations @@ -34,6 +35,7 @@ POLL = 25.0 NAME_LIMIT = 200 COMMAND_LIMIT = 4000 PATH_LIMIT = 4096 +LIMITS = {"tool": NAME_LIMIT, "command": COMMAND_LIMIT, "cwd": PATH_LIMIT, "session": NAME_LIMIT, "agent": NAME_LIMIT} DENIED = "The operator denied this in the kvb-hub." @@ -51,18 +53,12 @@ def facts(hook: dict) -> dict: "command": summary(hook.get("tool_input")), "cwd": str(hook.get("cwd") or ""), "session": str(hook.get("session_id") or ""), + "agent": str(hook.get("agent_type") or ""), } -def create(base: str, hook: dict, ask: dict, timeout: float) -> str: - payload = { - "tool": ask["tool"][:NAME_LIMIT], - "command": ask["command"][:COMMAND_LIMIT], - "cwd": ask["cwd"][:PATH_LIMIT], - "session": ask["session"][:NAME_LIMIT], - "agent": str(hook.get("agent_type") or "")[:NAME_LIMIT], - "timeout": timeout, - } +def create(base: str, ask: dict, timeout: float) -> str: + payload = {**ask, "timeout": timeout} headers = {"Content-Type": "application/json", **hub_api.GUARD} request = urllib.request.Request(f"{base}/api/asks", data=json.dumps(payload).encode(), headers=headers, method="POST") ask_id = hub_api.call(request, hub_api.SLACK)["id"] @@ -104,7 +100,11 @@ def main() -> int: try: hook = json.load(sys.stdin) ask = facts(hook) - ask_id = create(base, hook, ask, args.timeout) + too_long = [name for name, limit in LIMITS.items() if len(ask[name]) > limit] + if too_long: + print(f"kvb-hub: {', '.join(too_long)} too long to sign in full, Claude Code asks instead", file=sys.stderr) + return 0 + ask_id = create(base, ask, args.timeout) answer = {"status": "pending"} while answer["status"] == "pending" and (left := deadline - time.monotonic()) > 0: answer = status(base, ask_id, min(left, POLL)) diff --git a/hub/client/runner.py b/hub/client/runner.py index ccf5e79..a42052e 100755 --- a/hub/client/runner.py +++ b/hub/client/runner.py @@ -49,9 +49,11 @@ def call(base: str, path: str, payload: dict | None) -> dict: def collect(stream: object, tail: collections.deque, lock: threading.Lock) -> None: - for line in stream: - with lock: - tail.append(line) + # The reader closes the pipe at EOF: a close in execute() would block on the buffer lock this read holds and SIGPIPE a command still running. + with stream: + for line in stream: + with lock: + tail.append(line) def refusal(job: dict, trust: verifier.Trust, now: float) -> str | None: diff --git a/hub/kvb_hub/asking.py b/hub/kvb_hub/asking.py index 86d7643..bf041c4 100644 --- a/hub/kvb_hub/asking.py +++ b/hub/kvb_hub/asking.py @@ -55,7 +55,7 @@ class Asking: payload = decode(body) expect(payload, ("tool", *ASK_TEXTS, "timeout")) values = {name: bounded(payload, name, limit) for name, limit in ASK_TEXTS.items()} - lifetime = seconds(payload["timeout"], ASK_LIFETIME) + lifetime = seconds(payload, "timeout", ASK_LIFETIME) created = time.time() fields = {**values, "tool": field(payload, "tool").strip(), "status": "pending", "decided": None, "proof": None} record = self.asks.add({**fields, "created_ts": created, "expires": created + lifetime}) @@ -122,7 +122,7 @@ class Asking: def decision_wait(self, arguments: dict) -> dict: expect(arguments, ("id", "timeout_seconds")) decision_id = field(arguments, "id") - record = self.decisions.wait(decision_id, seconds(arguments["timeout_seconds"], WAIT_CAP)) + record = self.decisions.wait(decision_id, seconds(arguments, "timeout_seconds", WAIT_CAP)) if record is None: raise ToolError(f"no decision {decision_id}") return decision_state(record) diff --git a/hub/kvb_hub/exchange.py b/hub/kvb_hub/exchange.py index 4d4bb1c..4d7e8d7 100644 --- a/hub/kvb_hub/exchange.py +++ b/hub/kvb_hub/exchange.py @@ -97,7 +97,7 @@ class Exchange(BaseHTTPRequestHandler): def client(self) -> bool: return self.headers.get_all(GUARD) == ["1"] and self.headers.get_all("Origin") is None - def body(self, limit: int) -> bytes | None: + def body_length(self, limit: int) -> int | None: length = self.headers.get("Content-Length") if length is None or not LENGTH.fullmatch(length): self.send_text(HTTPStatus.LENGTH_REQUIRED, "Content-Length required") @@ -105,4 +105,4 @@ class Exchange(BaseHTTPRequestHandler): if int(length) > limit: self.send_text(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "body too large") return None - return self.rfile.read(int(length)) + return int(length) diff --git a/hub/kvb_hub/fields.py b/hub/kvb_hub/fields.py index dcda8a4..34e95f8 100644 --- a/hub/kvb_hub/fields.py +++ b/hub/kvb_hub/fields.py @@ -59,9 +59,10 @@ def message_text(arguments: dict, name: str) -> str: return value.strip() -def seconds(value: object, cap: float) -> float: +def seconds(arguments: dict, name: str, cap: float) -> float: + value = arguments[name] if isinstance(value, bool) or not isinstance(value, int | float) or not 0 <= value < math.inf: - raise ToolError("timeout_seconds must be a finite number of seconds, 0 or more") + raise ToolError(f"{name} must be a finite number of seconds, 0 or more") return float(min(value, cap)) diff --git a/hub/kvb_hub/handler.py b/hub/kvb_hub/handler.py index 90ec687..559d477 100644 --- a/hub/kvb_hub/handler.py +++ b/hub/kvb_hub/handler.py @@ -43,9 +43,6 @@ class Handler(Exchange): self.attempt(lambda: self.server.hub.speech.synthesize(body), lambda wav: self.send(HTTPStatus.OK, WAV_TYPE, wav, headers)) def hear(self, body: bytes) -> None: - if self.headers.get_content_type() != WAV_TYPE: - self.send_text(HTTPStatus.UNSUPPORTED_MEDIA_TYPE, f"use {WAV_TYPE}") - return self.answer(lambda: self.server.hub.speech.transcribe(body), HTTPStatus.OK) def do_GET(self) -> None: @@ -87,50 +84,50 @@ class Handler(Exchange): else: self.send_text(HTTPStatus.NOT_FOUND, "not found") - def action(self, path: str, body: bytes) -> Callable[[], None] | None: + def action(self, path: str) -> Callable[[bytes], None] | None: hub = self.server.hub route = ROUTE.fullmatch(path) item = ITEM.fullmatch(path) board = BOARD.fullmatch(path) if route and route.group(2) in DECISIONS: - return lambda: self.answer(lambda: hub.verdict(route.group(1), DECISIONS[route.group(2)], body), HTTPStatus.OK) + return lambda body: self.answer(lambda: hub.verdict(route.group(1), DECISIONS[route.group(2)], body), HTTPStatus.OK) if item and item.group(3): - return lambda: self.answer(lambda: hub.settle(item.group(1), item.group(2), item.group(3), body), HTTPStatus.OK) + return lambda body: self.answer(lambda: hub.settle(item.group(1), item.group(2), item.group(3), body), HTTPStatus.OK) if path == UPLOADS: - return lambda: self.answer(lambda: hub.upload(self.query(), body), HTTPStatus.CREATED) + return lambda body: self.answer(lambda: hub.upload(self.query(), body), HTTPStatus.CREATED) if path == "/api/messages": - return lambda: self.answer(lambda: hub.message(body), HTTPStatus.CREATED) + return lambda body: self.answer(lambda: hub.message(body), HTTPStatus.CREATED) if board: - return lambda: self.answer(lambda: hub.board(board.group(1), body), HTTPStatus.OK) - return self.inbox_action(path, body) + return lambda body: self.answer(lambda: hub.board(board.group(1), body), HTTPStatus.OK) + return self.inbox_action(path) - def inbox_action(self, path: str, body: bytes) -> Callable[[], None] | None: + def inbox_action(self, path: str) -> Callable[[bytes], None] | None: hub = self.server.hub inbox = INBOX_ROUTE.fullmatch(path) undo = UNDO_ROUTE.fullmatch(path) rule = RULE_ROUTE.fullmatch(path) if inbox and inbox.group(1) == "approve": - return lambda: self.answer(lambda: hub.approve(body), HTTPStatus.ACCEPTED) + return lambda body: self.answer(lambda: hub.approve(body), HTTPStatus.ACCEPTED) if path == SIGN_ROUTE: - return lambda: self.answer(lambda: hub.sign(body), HTTPStatus.OK) + return lambda body: self.answer(lambda: hub.sign(body), HTTPStatus.OK) if path == RULE_SIGN_ROUTE: - return lambda: self.answer(lambda: hub.sign_rule(body), HTTPStatus.OK) + return lambda body: self.answer(lambda: hub.sign_rule(body), HTTPStatus.OK) if inbox: - return lambda: self.answer(lambda: hub.mark(body), HTTPStatus.OK) + return lambda body: self.answer(lambda: hub.mark(body), HTTPStatus.OK) if undo: - return lambda: self.answer(lambda: hub.undo(undo.group(1)), HTTPStatus.OK) + return lambda _body: self.answer(lambda: hub.undo(undo.group(1)), HTTPStatus.OK) if path == "/api/rules": - return lambda: self.answer(lambda: hub.add_rule(body), HTTPStatus.CREATED) + return lambda body: self.answer(lambda: hub.add_rule(body), HTTPStatus.CREATED) if rule: - return lambda: self.answer(lambda: hub.rule_action(rule.group(1), rule.group(2)), HTTPStatus.OK) + return lambda _body: self.answer(lambda: hub.rule_action(rule.group(1), rule.group(2)), HTTPStatus.OK) if path == "/api/nl": - return lambda: self.answer(lambda: hub.nl(body), HTTPStatus.OK) + return lambda body: self.answer(lambda: hub.nl(body), HTTPStatus.OK) if path == SETTINGS_ROUTE: - return lambda: self.answer(lambda: hub.save_settings(body), HTTPStatus.OK) + return lambda body: self.answer(lambda: hub.save_settings(body), HTTPStatus.OK) if path == TTS_ROUTE: - return lambda: self.speak(body) + return self.speak if path == STT_ROUTE: - return lambda: self.hear(body) + return self.hear return None def posted(self, path: str) -> bool: @@ -139,41 +136,51 @@ class Handler(Exchange): def runner(self, path: str, body: bytes) -> None: hub = self.server.hub route = RUNNER_ROUTE.fullmatch(path) - if not self.client(): - self.send_text(HTTPStatus.FORBIDDEN, "forbidden: the runner calls with the X-Kvb-Hub header and no origin only") - elif route.group(2) == "claim": + if route.group(2) == "claim": self.answer(lambda: hub.claim(route.group(1)), HTTPStatus.OK) else: self.answer(lambda: hub.report(route.group(1), body), HTTPStatus.OK) + def receiver(self, path: str, host: str) -> Callable[[bytes], None] | None: + """Returns: + What handles the body once every header guard let the request + pass, or None after a guard answered it. + """ + if path == "/mcp": + return self.mcp_receiver(host) + if path == "/api/asks": + if self.local(host): + return lambda body: self.answer(lambda: self.server.hub.ask(body), HTTPStatus.CREATED) + self.send_text(HTTPStatus.FORBIDDEN, "forbidden: local POST with the X-Kvb-Hub header and no foreign origin only") + return None + if RUNNER_ROUTE.fullmatch(path): + if self.client(): + return lambda body: self.runner(path, body) + self.send_text(HTTPStatus.FORBIDDEN, "forbidden: the runner calls with the X-Kvb-Hub header and no origin only") + return None + action = self.action(path) + if action is None: + self.send_text(HTTPStatus.NOT_FOUND, "not found") + elif not self.guarded(host): + self.send_text(HTTPStatus.FORBIDDEN, "forbidden: same-origin POST with the X-Kvb-Hub header only") + elif path == STT_ROUTE and self.headers.get_content_type() != WAV_TYPE: + self.send_text(HTTPStatus.UNSUPPORTED_MEDIA_TYPE, f"use {WAV_TYPE}") + else: + return action + return None + def do_POST(self) -> None: path = urlsplit(self.path).path - body = self.body({UPLOADS: UPLOAD_LIMIT, STT_ROUTE: STT_LIMIT}.get(path, BODY_LIMIT)) - if body is None: + length = self.body_length({UPLOADS: UPLOAD_LIMIT, STT_ROUTE: STT_LIMIT}.get(path, BODY_LIMIT)) + if length is None: return host = self.host() if host is None: self.send_text(HTTPStatus.FORBIDDEN, "forbidden host") return - if path == "/mcp": - self.mcp(host, body) - return - if path == "/api/asks": - if self.local(host): - self.answer(lambda: self.server.hub.ask(body), HTTPStatus.CREATED) - else: - self.send_text(HTTPStatus.FORBIDDEN, "forbidden: local POST with the X-Kvb-Hub header and no foreign origin only") - return - if RUNNER_ROUTE.fullmatch(path): - self.runner(path, body) - return - action = self.action(path, body) - if action is None: - self.send_text(HTTPStatus.NOT_FOUND, "not found") - elif not self.guarded(host): - self.send_text(HTTPStatus.FORBIDDEN, "forbidden: same-origin POST with the X-Kvb-Hub header only") - else: - action() + receive = self.receiver(path, host) + if receive is not None: + receive(self.rfile.read(length)) def index(self) -> None: nonce = secrets.token_urlsafe(18) @@ -188,16 +195,19 @@ class Handler(Exchange): if reviews.get(review_id) is None or not path.is_file(): self.send_text(HTTPStatus.NOT_FOUND, "not found") return - self.send(HTTPStatus.OK, content_type, path.read_bytes(), {"Content-Security-Policy": csp}) + self.attempt(path.read_bytes, lambda data: self.send(HTTPStatus.OK, content_type, data, {"Content-Security-Policy": csp})) - def mcp(self, host: str, body: bytes) -> None: + def mcp_receiver(self, host: str) -> Callable[[bytes], None] | None: origins = self.headers.get_all("Origin") if origins is not None and origins != [f"http://{host}"]: self.send_text(HTTPStatus.FORBIDDEN, "foreign origin") - return - if self.headers.get_content_type() != "application/json": + elif self.headers.get_content_type() != "application/json": self.send_text(HTTPStatus.UNSUPPORTED_MEDIA_TYPE, "use application/json") - return + else: + return self.mcp + return None + + def mcp(self, body: bytes) -> None: status, reply = self.server.hub.rpc(body) if reply is not None: self.send_json(status, reply) diff --git a/hub/kvb_hub/language.py b/hub/kvb_hub/language.py index 9204ba8..1397553 100644 --- a/hub/kvb_hub/language.py +++ b/hub/kvb_hub/language.py @@ -180,7 +180,9 @@ class Language: now = time.time() claims = read_claims(self.council) tasks = read_register(self.council / REGISTER) - items = self.inbox(now, tasks, self.directory(claims, read_jsonl(self.council / NEWS), now)) + directory = self.directory(claims, read_jsonl(self.council / NEWS), now) + self.watch.see(tasks, now) + items = self.inbox(now, tasks, directory) agents = ordered({claim["agent"] for claim in claims if claim["agent"]} | set(priority(tasks))) shown_items = [{"id": item["id"], "kind": item["kind"], "title": item["title"][:200]} for item in items] return nl_actions(self.model.complete(nl_messages(text, shown_items, agents)), items, agents) diff --git a/hub/kvb_hub/reviewing.py b/hub/kvb_hub/reviewing.py index f06a2bb..0cb6454 100644 --- a/hub/kvb_hub/reviewing.py +++ b/hub/kvb_hub/reviewing.py @@ -97,7 +97,7 @@ class Reviewing: def wait(self, arguments: dict) -> dict: expect(arguments, ("id", "timeout_seconds")) review_id = field(arguments, "id") - timeout = seconds(arguments["timeout_seconds"], WAIT_CAP) + timeout = seconds(arguments, "timeout_seconds", WAIT_CAP) return state_of(found(self.reviews.wait(review_id, timeout), review_id)) def pending(self, arguments: dict) -> dict: diff --git a/hub/templates/index.html.j2 b/hub/templates/index.html.j2 index 9cbb5c0..98cded3 100644 --- a/hub/templates/index.html.j2 +++ b/hub/templates/index.html.j2 @@ -4,7 +4,7 @@ -{% if state.inbox %}({{ state.inbox | length }}) {% endif %}kvb-hub +{% set waiting = state.inbox | rejectattr("until") | list | length %}{% if waiting %}({{ waiting }}) {% endif %}kvb-hub