wip(13-02): checkpoint interrupted model-proxy work (session-gated forwarder)
Session died on a broken pipe with this work uncommitted in the executor worktree. Committed verbatim, unverified — not a task completion. The continuation executor may reset --soft this commit and recommit atomically per task. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
archipelago
co-authored by
Claude Opus 5
Claude Opus 5
parent
15774d266f
commit
13b576da26
@@ -1,6 +1,7 @@
|
||||
mod blob;
|
||||
mod content;
|
||||
mod dwn;
|
||||
mod model_proxy;
|
||||
mod node_message;
|
||||
mod proxy;
|
||||
mod remote_input;
|
||||
@@ -433,6 +434,17 @@ impl ApiHandler {
|
||||
// RPC — auth is handled inside rpc handler per-method
|
||||
(Method::POST, "/rpc/v1") => self.rpc_handler.clone().handle(req_with_bytes).await,
|
||||
|
||||
// AIUI model proxy — session-gated forwarder to Claude/Ollama,
|
||||
// replacing the unauthenticated claude-api-proxy.py sidecar and
|
||||
// the /aiui/api/openrouter/ open relay (13-02-PLAN.md,
|
||||
// T-13-08/T-13-09/T-13-10/T-13-11). The daemon re-derives auth
|
||||
// from the cookie inside handle_model_proxy — it does not trust
|
||||
// nginx to have gated the request already, the same "don't trust
|
||||
// the front door" discipline as /lnd-connect-info below.
|
||||
(_, p) if p.starts_with("/aiui/api/claude/") || p.starts_with("/aiui/api/ollama/") => {
|
||||
self.handle_model_proxy(req_with_bytes, p).await
|
||||
}
|
||||
|
||||
// Health — unauthenticated, returns JSON with service status
|
||||
(Method::GET, "/health") => {
|
||||
let recovery_complete = crate::crash_recovery::is_recovery_complete();
|
||||
|
||||
@@ -0,0 +1,400 @@
|
||||
//! Session-gated forwarder for `/aiui/api/claude/*` and `/aiui/api/ollama/*`.
|
||||
//!
|
||||
//! Replaces `claude-api-proxy.py` — a standalone Python process on port 3142
|
||||
//! holding its **own** copy of the Anthropic API key, reachable with **no
|
||||
//! session gate** — and retires the `/aiui/api/openrouter/` open relay
|
||||
//! entirely (13-02-PLAN.md, T-13-08/T-13-09/T-13-10/T-13-11/T-13-12). Anyone
|
||||
//! who could reach the node's web port could spend the owner's API budget.
|
||||
//!
|
||||
//! The daemon re-derives auth from the request's own session cookie — it
|
||||
//! does not trust nginx to have gated the request already, the same
|
||||
//! discipline `/lnd-connect-info`'s doc comment spells out for exactly this
|
||||
//! reason (a second front door, or a misconfigured proxy, must not become a
|
||||
//! silent bypass). It reads the node's single Claude key ledger
|
||||
//! (`data_dir/secrets/claude-api-key`) fresh on every call rather than
|
||||
//! caching it, and never forwards an inbound `x-api-key`, `authorization`
|
||||
//! or `cookie` header upstream (T-13-14) — a caller must not be able to
|
||||
//! bill a different account or leak the node's session to Anthropic.
|
||||
|
||||
use super::ApiHandler;
|
||||
use crate::session::{self, SessionStore};
|
||||
use anyhow::Result;
|
||||
use hyper::{Body, HeaderMap, Method, Request, Response, StatusCode};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::Duration;
|
||||
|
||||
/// Anthropic Messages API base. The node's single key ledger
|
||||
/// (`data_dir/secrets/claude-api-key`) authenticates every forwarded call.
|
||||
const CLAUDE_UPSTREAM: &str = "https://api.anthropic.com/";
|
||||
/// Local Ollama. No key — the session gate exists purely to stop anonymous
|
||||
/// consumption of local GPU/CPU inference (T-13-11), not to protect a secret.
|
||||
const OLLAMA_UPSTREAM: &str = "http://127.0.0.1:11434/";
|
||||
/// Generous enough for a multi-turn tool-call round trip; `mesh/listener/
|
||||
/// assist.rs`'s OLLAMA_TIMEOUT (60s) is airtime-tuned for LoRa and not
|
||||
/// reusable here — this path has no such constraint (13-AI-SPEC.md Pitfall 6).
|
||||
const FORWARD_TIMEOUT_SECS: u64 = 180;
|
||||
|
||||
impl ApiHandler {
|
||||
/// Entry point wired into the `/aiui/api/claude/` and `/aiui/api/ollama/`
|
||||
/// arms in `mod.rs`. Kept as a thin method so it can read
|
||||
/// `self.session_store` / `self.config.data_dir`; the actual routing and
|
||||
/// forwarding logic lives in free functions below so it is unit-testable
|
||||
/// without constructing a full `ApiHandler` (RpcHandler + orchestrators +
|
||||
/// blob store) in every test.
|
||||
pub(super) async fn handle_model_proxy(
|
||||
&self,
|
||||
req: Request<Body>,
|
||||
path: &str,
|
||||
) -> Result<Response<Body>> {
|
||||
route_model_proxy(&self.session_store, &self.config.data_dir, req, path).await
|
||||
}
|
||||
}
|
||||
|
||||
/// Routing + auth gate, factored out of the `ApiHandler` method so tests can
|
||||
/// exercise it with `SessionStore::new_for_tests` and a `tempfile` data_dir.
|
||||
async fn route_model_proxy(
|
||||
session_store: &SessionStore,
|
||||
data_dir: &Path,
|
||||
req: Request<Body>,
|
||||
path: &str,
|
||||
) -> Result<Response<Body>> {
|
||||
if !is_authenticated(session_store, req.headers()).await {
|
||||
tracing::warn!("401 model proxy {} — session invalid or missing", path);
|
||||
return Ok(unauthorized());
|
||||
}
|
||||
if let Some(rest) = path.strip_prefix("/aiui/api/claude/") {
|
||||
forward_claude(req, rest, data_dir).await
|
||||
} else if let Some(rest) = path.strip_prefix("/aiui/api/ollama/") {
|
||||
forward_ollama(req, rest).await
|
||||
} else {
|
||||
// Unreachable given the caller's prefix match in mod.rs, but never
|
||||
// fall through to an unauthenticated 200 on an unrecognized path.
|
||||
Ok(unauthorized())
|
||||
}
|
||||
}
|
||||
|
||||
/// Re-derive session auth from the request's own cookie. Deliberately not a
|
||||
/// call back into `ApiHandler::is_authenticated` — keeping this small and
|
||||
/// dependency-free is what makes the 401 behaviour unit-testable without
|
||||
/// paying for a full `ApiHandler` in every test.
|
||||
async fn is_authenticated(session_store: &SessionStore, headers: &HeaderMap) -> bool {
|
||||
match session::extract_session_cookie(headers) {
|
||||
Some(token) => session_store.validate(&token).await,
|
||||
None => false,
|
||||
}
|
||||
}
|
||||
|
||||
fn unauthorized() -> Response<Body> {
|
||||
let body = serde_json::json!({ "error": "Unauthorized" });
|
||||
Response::builder()
|
||||
.status(StatusCode::UNAUTHORIZED)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(Body::from(serde_json::to_vec(&body).unwrap_or_default()))
|
||||
.unwrap_or_else(|_| Response::new(Body::from("Unauthorized")))
|
||||
}
|
||||
|
||||
/// A plain-language 503 naming the missing key — never a 500, and never the
|
||||
/// key's filesystem path (that would hand an authenticated-but-untrusted
|
||||
/// caller a hint about the node's on-disk layout for no benefit to them).
|
||||
fn key_not_configured() -> Response<Body> {
|
||||
let body = serde_json::json!({
|
||||
"error": "Claude is not configured on this node yet — set an API key in Settings."
|
||||
});
|
||||
Response::builder()
|
||||
.status(StatusCode::SERVICE_UNAVAILABLE)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(Body::from(serde_json::to_vec(&body).unwrap_or_default()))
|
||||
.unwrap_or_else(|_| Response::new(Body::from("Claude is not configured")))
|
||||
}
|
||||
|
||||
fn bad_gateway(msg: &str) -> Response<Body> {
|
||||
let body = serde_json::json!({ "error": msg });
|
||||
Response::builder()
|
||||
.status(StatusCode::BAD_GATEWAY)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(Body::from(serde_json::to_vec(&body).unwrap_or_default()))
|
||||
.unwrap_or_else(|_| Response::new(Body::from(msg.to_string())))
|
||||
}
|
||||
|
||||
/// Forward an already-authenticated request to Anthropic's Messages API.
|
||||
/// `rest` is the path remainder after `/aiui/api/claude/` has been stripped
|
||||
/// by the caller (e.g. `v1/messages`).
|
||||
async fn forward_claude(req: Request<Body>, rest: &str, data_dir: &Path) -> Result<Response<Body>> {
|
||||
let key_path: PathBuf = data_dir.join("secrets/claude-api-key");
|
||||
let api_key = match tokio::fs::read_to_string(&key_path).await {
|
||||
Ok(k) if !k.trim().is_empty() => k.trim().to_string(),
|
||||
_ => {
|
||||
tracing::warn!("model proxy: claude key ledger missing, refusing forward");
|
||||
return Ok(key_not_configured());
|
||||
}
|
||||
};
|
||||
forward(
|
||||
req,
|
||||
rest,
|
||||
CLAUDE_UPSTREAM,
|
||||
"api.anthropic.com",
|
||||
&[
|
||||
("x-api-key", api_key),
|
||||
("anthropic-version", "2023-06-01".to_string()),
|
||||
],
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Forward an already-authenticated request to the node's local Ollama.
|
||||
/// `rest` is the path remainder after `/aiui/api/ollama/` has been stripped.
|
||||
async fn forward_ollama(req: Request<Body>, rest: &str) -> Result<Response<Body>> {
|
||||
forward(req, rest, OLLAMA_UPSTREAM, "127.0.0.1:11434", &[]).await
|
||||
}
|
||||
|
||||
/// Shared forwarding core for both backends. Copies ONLY the inbound
|
||||
/// `content-type`/`accept` request headers plus whatever `extra_headers`
|
||||
/// the caller supplies (the Claude key + version pin) — the inbound
|
||||
/// `x-api-key`, `authorization` and `cookie` headers are never read, let
|
||||
/// alone forwarded (T-13-14). Streams the upstream response back rather
|
||||
/// than buffering it, matching `proxy.rs`'s peer-content streaming shape,
|
||||
/// so token-by-token replies still stream to the browser.
|
||||
async fn forward(
|
||||
req: Request<Body>,
|
||||
rest: &str,
|
||||
upstream_base: &str,
|
||||
upstream_host_for_log: &str,
|
||||
extra_headers: &[(&str, String)],
|
||||
) -> Result<Response<Body>> {
|
||||
let method = req.method().clone();
|
||||
let (parts, body) = req.into_parts();
|
||||
let content_type = parts
|
||||
.headers
|
||||
.get(hyper::header::CONTENT_TYPE)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("application/json")
|
||||
.to_string();
|
||||
let accept = parts
|
||||
.headers
|
||||
.get(hyper::header::ACCEPT)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map(|s| s.to_string());
|
||||
let payload = hyper::body::to_bytes(body)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("read request payload: {e}"))?;
|
||||
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(FORWARD_TIMEOUT_SECS))
|
||||
.build()
|
||||
.map_err(|e| anyhow::anyhow!("client build: {e}"))?;
|
||||
|
||||
let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())
|
||||
.unwrap_or(reqwest::Method::POST);
|
||||
let url = format!("{}{}", upstream_base, rest);
|
||||
let mut upstream_req = client
|
||||
.request(reqwest_method, &url)
|
||||
.header("content-type", content_type);
|
||||
for (name, value) in extra_headers {
|
||||
upstream_req = upstream_req.header(*name, value);
|
||||
}
|
||||
if let Some(accept) = accept {
|
||||
upstream_req = upstream_req.header("accept", accept);
|
||||
}
|
||||
// GET requests to Ollama carry no payload; avoid sending an empty body
|
||||
// on GET, which some servers treat differently from "no body at all".
|
||||
if method != Method::GET || !payload.is_empty() {
|
||||
upstream_req = upstream_req.body(payload.to_vec());
|
||||
}
|
||||
|
||||
match upstream_req.send().await {
|
||||
Ok(resp) => {
|
||||
let status = resp.status().as_u16();
|
||||
tracing::info!(
|
||||
"model proxy: forwarded to {}, status={}",
|
||||
upstream_host_for_log,
|
||||
status
|
||||
);
|
||||
stream_response(resp)
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"model proxy: upstream request to {} failed: {}",
|
||||
upstream_host_for_log,
|
||||
e
|
||||
);
|
||||
Ok(bad_gateway("upstream request failed"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Stream the upstream response straight through instead of buffering it —
|
||||
/// same shape as `proxy.rs`'s peer-content Range streamer — so a
|
||||
/// token-by-token reply doesn't wait for the full response before the first
|
||||
/// byte reaches the browser.
|
||||
fn stream_response(resp: reqwest::Response) -> Result<Response<Body>> {
|
||||
let status = resp.status().as_u16();
|
||||
let headers = resp.headers().clone();
|
||||
let mut builder = Response::builder().status(status);
|
||||
for h in ["content-type", "content-length"] {
|
||||
if let Some(v) = headers.get(h).and_then(|v| v.to_str().ok()) {
|
||||
builder = builder.header(h, v);
|
||||
}
|
||||
}
|
||||
builder
|
||||
.body(Body::wrap_stream(resp.bytes_stream()))
|
||||
.map_err(|e| anyhow::anyhow!("response build: {e}"))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::Mutex as TokioMutex;
|
||||
|
||||
/// Unique suffix for a per-test temp file path (matches the pattern
|
||||
/// `session.rs`'s own tests already use — not key material, just a
|
||||
/// filename component, drawn unguarded).
|
||||
fn uniq() -> u64 {
|
||||
rand::RngCore::next_u64(&mut rand::rngs::OsRng)
|
||||
}
|
||||
|
||||
async fn test_store() -> SessionStore {
|
||||
let dir = std::env::temp_dir();
|
||||
let path = dir.join(format!("archy-model-proxy-test-sessions-{}.json", uniq()));
|
||||
SessionStore::new_for_tests(path)
|
||||
}
|
||||
|
||||
fn req_with_cookie(method: &str, path: &str, cookie: Option<&str>) -> Request<Body> {
|
||||
let mut builder = Request::builder().method(method).uri(path);
|
||||
if let Some(c) = cookie {
|
||||
builder = builder.header("cookie", format!("session={c}"));
|
||||
}
|
||||
builder.body(Body::empty()).unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn claude_without_session_is_401() {
|
||||
let store = test_store().await;
|
||||
let data_dir = tempfile::tempdir().unwrap();
|
||||
let req = req_with_cookie("POST", "/aiui/api/claude/v1/messages", None);
|
||||
let resp = route_model_proxy(
|
||||
&store,
|
||||
data_dir.path(),
|
||||
req,
|
||||
"/aiui/api/claude/v1/messages",
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ollama_without_session_is_401() {
|
||||
let store = test_store().await;
|
||||
let data_dir = tempfile::tempdir().unwrap();
|
||||
let req = req_with_cookie("GET", "/aiui/api/ollama/api/tags", None);
|
||||
let resp = route_model_proxy(&store, data_dir.path(), req, "/aiui/api/ollama/api/tags")
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn claude_with_invalid_session_is_401() {
|
||||
let store = test_store().await;
|
||||
let data_dir = tempfile::tempdir().unwrap();
|
||||
let req = req_with_cookie(
|
||||
"POST",
|
||||
"/aiui/api/claude/v1/messages",
|
||||
Some("not-a-real-token"),
|
||||
);
|
||||
let resp = route_model_proxy(
|
||||
&store,
|
||||
data_dir.path(),
|
||||
req,
|
||||
"/aiui/api/claude/v1/messages",
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn missing_key_is_503_not_500() {
|
||||
let store = test_store().await;
|
||||
let token = store.create().await;
|
||||
// Deliberately no data_dir/secrets/claude-api-key written.
|
||||
let data_dir = tempfile::tempdir().unwrap();
|
||||
let req = req_with_cookie("POST", "/aiui/api/claude/v1/messages", Some(&token));
|
||||
let resp = route_model_proxy(
|
||||
&store,
|
||||
data_dir.path(),
|
||||
req,
|
||||
"/aiui/api/claude/v1/messages",
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
|
||||
}
|
||||
|
||||
/// Minimal local capture server (hyper 0.14, same crate `server.rs`
|
||||
/// already builds on) standing in for an upstream — records the headers
|
||||
/// of the one request it receives so the test can assert what actually
|
||||
/// left the node, without adding a mocking dependency.
|
||||
async fn spawn_capture_server() -> (String, Arc<TokioMutex<Option<HeaderMap>>>) {
|
||||
let captured: Arc<TokioMutex<Option<HeaderMap>>> = Arc::new(TokioMutex::new(None));
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let captured_clone = captured.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Ok((stream, _)) = listener.accept().await {
|
||||
let captured = captured_clone.clone();
|
||||
let service = hyper::service::service_fn(move |req: Request<Body>| {
|
||||
let captured = captured.clone();
|
||||
async move {
|
||||
*captured.lock().await = Some(req.headers().clone());
|
||||
Ok::<_, std::convert::Infallible>(Response::new(Body::from("{}")))
|
||||
}
|
||||
});
|
||||
let _ = hyper::server::conn::Http::new()
|
||||
.serve_connection(stream, service)
|
||||
.await;
|
||||
}
|
||||
});
|
||||
(format!("http://{addr}/"), captured)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn inbound_authorization_header_is_not_forwarded() {
|
||||
let (upstream, captured) = spawn_capture_server().await;
|
||||
let req = Request::builder()
|
||||
.method("POST")
|
||||
.uri("/v1/messages")
|
||||
.header("authorization", "Bearer caller-supplied-secret")
|
||||
.header("x-api-key", "attacker-supplied-key")
|
||||
.header("cookie", "session=some-session-token")
|
||||
.header("content-type", "application/json")
|
||||
.body(Body::from("{}"))
|
||||
.unwrap();
|
||||
let resp = forward(req, "v1/messages", &upstream, "test-upstream", &[])
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(resp.status().is_success());
|
||||
|
||||
// Give the spawned capture task a moment to record the request.
|
||||
for _ in 0..20 {
|
||||
if captured.lock().await.is_some() {
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
}
|
||||
let headers = captured
|
||||
.lock()
|
||||
.await
|
||||
.clone()
|
||||
.expect("capture server did not receive a request");
|
||||
assert!(headers.get("authorization").is_none());
|
||||
assert!(headers.get("x-api-key").is_none());
|
||||
assert!(headers.get("cookie").is_none());
|
||||
// The one header we DO expect to survive the round trip.
|
||||
assert_eq!(
|
||||
headers.get("content-type").and_then(|v| v.to_str().ok()),
|
||||
Some("application/json")
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -1049,22 +1049,12 @@ impl RpcHandler {
|
||||
info!("Claude API key saved");
|
||||
}
|
||||
|
||||
// Update the claude-api-proxy environment and restart
|
||||
let env_line = format!("ANTHROPIC_API_KEY={}", value);
|
||||
let env_file = self.config.data_dir.join("secrets/claude-api-proxy.env");
|
||||
tokio::fs::write(&env_file, &env_line).await.ok();
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
std::fs::set_permissions(&env_file, std::fs::Permissions::from_mode(0o600))
|
||||
.ok();
|
||||
}
|
||||
|
||||
// Restart the proxy to pick up the new key
|
||||
let _ = tokio::process::Command::new("sudo")
|
||||
.args(["systemctl", "restart", "claude-api-proxy"])
|
||||
.output()
|
||||
.await;
|
||||
// `secrets/claude-api-key` (above) is deliberately the ONLY
|
||||
// Claude key ledger on this node (13-02-PLAN.md). A second
|
||||
// copy used to be written alongside it for a standalone,
|
||||
// unauthenticated sidecar process on port 3142 — that
|
||||
// sidecar and its key copy are retired; the session-gated
|
||||
// Rust daemon reads this one file directly.
|
||||
|
||||
Ok(serde_json::json!({ "saved": true }))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user