Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150"""Signed webhook ingress wakes the same durable session; no live provider."""import contextlibimport hashlibimport hmacimport http.serverimport jsonimport osfrom pathlib import Pathimport socketimport subprocessimport tempfileimport threadingimport timeimport urllib.errorimport urllib.request
ROOT = Path(__file__).resolve().parents[2]
class Provider(http.server.BaseHTTPRequestHandler): requests = []
def log_message(self, *_): pass
def do_POST(self): request = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) self.requests.append(request) output = [{"type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "acknowledged", "annotations": []}]}] if (any(item.get("role") == "user" and "probe webhook binding" in str(item.get("content", "")) for item in request["input"]) and request["input"][-1].get("type") != "function_call_output"): code = ("hooks = await webhooks.list()\n" "assert any(h.name == 'outage' for h in hooks)\n" f"payload = await webhooks.delivery('{self.delivery_id}')\n" "assert 'down' in payload.body\n" "print('WEBHOOK_BINDING_OK')") output = [{"type": "function_call", "id": "fc-hook", "call_id": "call-hook", "name": "python", "arguments": json.dumps({"code": code, "timeout_ms": 10000}), "status": "completed"}] event = {"type": "response.completed", "response": { "id": "fixture", "status": "completed", "output": output, "usage": {"input_tokens": 20, "output_tokens": 2}, }} body = ("data: " + json.dumps(event) + "\n\n").encode() self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body)
def free_port(): with socket.socket() as probe: probe.bind(("127.0.0.1", 0)) return probe.getsockname()[1]
def status(call): try: with call() as response: return response.status except urllib.error.HTTPError as error: return error.code
def main(): provider = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Provider) threading.Thread(target=provider.serve_forever, daemon=True).start() with tempfile.TemporaryDirectory(prefix="albedo-webhooks-") as directory: root = Path(directory) home, workspace, user_home = root/"state", root/"workspace", root/"user" for path in (home, workspace, user_home): path.mkdir(mode=0o700) (home/"config.json").write_text(json.dumps({"active": "fixture", "providers": {"fixture": { "baseUrl": f"http://127.0.0.1:{provider.server_address[1]}/v1", "apiKey": "key", "model": "fixture", "protocol": "responses", }}})) (home/"extensions.json").write_text(json.dumps({"models": {"refreshHours": 0}})) port = free_port() env = dict(os.environ, HOME=str(user_home), ALBEDO_HOME=str(home), ALBEDO_PORT=str(port), ALBEDO_PARENT_PID=str(os.getpid()), ALBEDO_NO_BROWSER="1") connection = None
def cli(*args): result = subprocess.run([str(ROOT/"cli/bin/albedo"), *args], cwd=ROOT, env=env, capture_output=True, text=True, timeout=45) assert result.returncode == 0, result.stdout + result.stderr + (home/"daemon.log").read_text() return result.stdout
def api(path, data=None): request = urllib.request.Request(f"http://127.0.0.1:{port}" + path, headers={"Authorization": "Bearer " + connection["token"], "Content-Type": "application/json"}, data=None if data is None else json.dumps(data).encode()) with urllib.request.urlopen(request, timeout=25) as response: return json.load(response)
try: cli("sessions") connection = json.loads((home/"daemon.json").read_text()) session = json.loads(cli("new", str(workspace)))["session"] api(f"/sessions/{session}/extensions", {"name": "webhooks", "scope": "global", "enabled": True}) command = f"/sessions/{session}/commands" created = api(command, {"name": "/webhooks", "args": {"action": "create", "details": "outage"}})["result"] hook, secret = created["hook"], created["secret"] url = f"http://127.0.0.1:{port}" + hook["url"] payload = b'{"status":"down"}' signature = "sha256=" + hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
def send(body=payload, sig=signature, event_id="alert-1", origin=None): headers = {"X-Albedo-Signature": sig, "X-Albedo-Event-Id": event_id} if origin: headers["Origin"] = origin return urllib.request.urlopen(urllib.request.Request(url, data=body, headers=headers), timeout=25)
assert status(lambda: send(sig="sha256=wrong")) == 401 assert status(lambda: send(origin="https://attacker.example")) == 403 with send() as response: delivery = json.load(response)["deliveryId"] assert response.status == 202 with send() as response: assert json.load(response)["deliveryId"] == delivery assert status(lambda: send(body=b"different", sig="sha256=" + hmac.new(secret.encode(), b"different", hashlib.sha256).hexdigest())) == 409 deadline = time.monotonic() + 40 while time.monotonic() < deadline and not Provider.requests: time.sleep(.25) assert Provider.requests, (home/"daemon.log").read_text() assert any("[webhook outage #" in json.dumps(request) for request in Provider.requests) assert len(Provider.requests) == 1, "retry should wake the session once" assert api(command, {"name": "/webhooks", "args": {"action": "list"}})["result"]["hooks"][0]["hook"]["id"] == hook["id"] api(command, {"name": "/webhooks", "args": {"action": "agent_on"}}) Provider.delivery_id = delivery api(f"/sessions/{session}/events", {"content": "probe webhook binding"}) deadline = time.monotonic() + 20 while time.monotonic() < deadline and len(Provider.requests) < 3: time.sleep(.1) assert len(Provider.requests) == 3, (home/"daemon.log").read_text() tool_outputs = [item.get("output", "") for item in Provider.requests[-1]["input"] if item.get("type") == "function_call_output"] assert any("WEBHOOK_BINDING_OK" in str(output) for output in tool_outputs), tool_outputs finally: if connection: with contextlib.suppress(Exception): api("/shutdown", {}) provider.shutdown()
if __name__ == "__main__": main()