Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -49,12 +49,21 @@ ClawShell supports OAuth-based authentication as an alternative to static API ke
- **Automatic Token Refresh**: Access tokens are refreshed transparently before they expire.
- **Request Translation**: Automatically translates OpenAI Chat Completions API requests to the ChatGPT Responses API format when using Codex OAuth.

### 5. Seamless Integration
### 5. Runtime Statistics

ClawShell exposes running counters at `GET /admin/stats` so operators can audit proxy activity since startup and across restarts.

- **What's Counted**: Total requests served, total upstream `prompt_tokens` / `completion_tokens` / `total_tokens` (parsed from non-streaming JSON responses — SSE streams are not counted), number of emails hidden by the sender policy, and a per-address breakdown of filtered senders.
- **Loopback-Only**: The endpoint is reachable without a virtual key but only from `127.0.0.1` / `::1` peers; non-loopback clients receive `403`.
- **Persistent**: Counters are flushed to disk every 30 seconds and on graceful shutdown. The location is a required config field — set `[stats] persist_path = "..."` in `clawshell.toml` (typically `/var/lib/clawshell/stats.json` under the hardened systemd unit, since `/etc/clawshell` is read-only there).
- **Bounded**: The filtered-address map is capped at 10,000 unique entries; further unique addresses are aggregated under an `<overflow>` bucket so memory stays bounded.

### 6. Seamless Integration

- **Drop-in Sidecar**: The `clawshell onboard` wizard configures exactly one downstream LLM client per run — either OpenClaw or [Hermes Agent](https://github.com/NousResearch/hermes-agent) — to route all requests through ClawShell's proxy. See [Agent Target (pick one)](#agent-target-pick-one).
- **No External Dependencies**: Uses Unix file system permissions to protect secrets. No IdP, Vault, or external key management service required.

### 6. Ultra Lightweight and Scalable
### 7. Ultra Lightweight and Scalable

- Runs in under 10MB of memory.
- Written in Rust with Tokio.
Expand Down
7 changes: 7 additions & 0 deletions clawshell.example.toml
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,13 @@ patterns = [
{ name = "amex_card", regex = '\b3[47][0-9]{13}\b', action = "redact" },
]

# Runtime statistics persistence
# ClawShell counts total requests served, upstream prompt/completion/total
# tokens (from non-streaming responses), and per-sender email-filter
# activity, and exposes them at GET /admin/stats (loopback-only).
[stats]
persist_path = "/etc/clawshell/stats.json"

# Email read endpoint
# If enabled, set exactly one mode:
# - mode = "allowlist" with non-empty allow_senders and empty deny_senders
Expand Down
128 changes: 88 additions & 40 deletions src/app.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,11 @@ use crate::email::{
use crate::keys::{KeyManager, KeySource, ResolvedKey};
use crate::oauth::OAuthRegistry;
use crate::proxy::ProxyClient;
use crate::stats::Stats;

use axum::Router;
use axum::body::Body;
use axum::extract::{DefaultBodyLimit, Path, Query, Request, State};
use axum::extract::{ConnectInfo, DefaultBodyLimit, Path, Query, Request, State};
use axum::http::{HeaderMap, Method, StatusCode, Uri};
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
Expand All @@ -20,6 +21,7 @@ use bytes::Bytes;
use http_body_util::BodyExt;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Instant;
use tracing::{debug, error, info, trace, warn};
Expand All @@ -34,6 +36,7 @@ pub struct AppState {
pub email_policy: Option<EmailPolicy>,
pub email_accounts: Arc<BTreeMap<String, EmailAccountCredentials>>,
pub email_service: Arc<EmailService>,
pub stats: Arc<Stats>,
}

impl AppState {
Expand Down Expand Up @@ -138,6 +141,7 @@ impl AppState {
email_policy,
email_accounts: Arc::new(email_accounts),
email_service: Arc::new(EmailService::Imap(ImapEmailService::default())),
stats: Arc::new(Stats::new(Some(config.stats.persist_path.clone()))),
})
}
}
Expand All @@ -149,20 +153,30 @@ pub fn build_router(state: AppState) -> Router {
Router::new()
.route("/v1/email/messages", get(handle_email_secure_messages))
.route("/v1/email/messages/{id}", get(handle_email_message_content))
.route("/admin/stats", get(handle_stats))
.route("/", any(handle_request))
.route("/{*path}", any(handle_request))
.layer(DefaultBodyLimit::max(MAX_BODY_SIZE))
.layer(axum::middleware::from_fn(log_request_completion))
.layer(axum::middleware::from_fn_with_state(
state.clone(),
log_request_completion,
))
.with_state(state)
}

async fn log_request_completion(request: Request, next: Next) -> Response {
async fn log_request_completion(
State(state): State<AppState>,
request: Request,
next: Next,
) -> Response {
let start = Instant::now();
let method = request.method().clone();
let path = request.uri().path().to_string();
let query = request.uri().query().map(|q| q.to_string());
let response = next.run(request).await;

state.stats.record_request();

info!(
method = %method,
path = %path,
Expand Down Expand Up @@ -317,6 +331,7 @@ async fn handle_email_secure_messages(
continue;
};
if !policy.sender_visible(from_header) {
state.stats.record_email_filtered(from_header);
continue;
}
visible_messages.push(EmailSecureMessage {
Expand Down Expand Up @@ -447,6 +462,9 @@ async fn handle_email_message_content(
.as_deref()
.or_else(|| content.headers.get("from").map(String::as_str));
if from_header.is_none_or(|from| !policy.sender_visible(from)) {
if let Some(from) = from_header {
state.stats.record_email_filtered(from);
}
return Err(error_response(
StatusCode::NOT_FOUND,
"Email message not found",
Expand All @@ -469,6 +487,27 @@ async fn handle_email_message_content(
Ok(axum::Json(response).into_response())
}

/// Management endpoint that returns running counters for total requests,
/// upstream token usage, and per-sender email-filter activity. Only
/// reachable from loopback peers; streaming (SSE) responses are not
/// included in token totals (see `handle_request`).
async fn handle_stats(
State(state): State<AppState>,
ConnectInfo(peer): ConnectInfo<SocketAddr>,
) -> Result<Response, Response> {
if !peer.ip().is_loopback() {
warn!(
peer = %peer,
"Non-loopback client tried to hit /admin/stats"
);
return Err(error_response(
StatusCode::FORBIDDEN,
"stats endpoint is loopback-only",
));
}
Ok(axum::Json(state.stats.snapshot()).into_response())
}

async fn handle_request(
State(state): State<AppState>,
request: Request,
Expand Down Expand Up @@ -630,31 +669,52 @@ async fn handle_request(
})?,
};

// 5. DLP scan on response body (redact all PII before returning to client)
let response = if state.dlp_scanner.scan_responses() {
trace!("Response DLP scanning enabled, checking response body");
let is_streaming = response
.headers()
.get("content-type")
.and_then(|v| v.to_str().ok())
.is_some_and(|ct| ct.contains("text/event-stream"));
// 5. Response processing.
// Non-streaming responses are buffered so we can (a) record upstream
// token usage for stats and (b) optionally DLP-redact before sending.
// SSE streams are handled by the existing DLP SSE wrapper; we do not
// count tokens from SSE usage events.
let is_streaming = response
.headers()
.get("content-type")
.and_then(|v| v.to_str().ok())
.is_some_and(|ct| ct.contains("text/event-stream"));

if !is_streaming {
trace!("Non-streaming response, scanning body for PII");
let response = if is_streaming {
if state.dlp_scanner.scan_responses() {
debug!(
method = %method,
path = %path,
virtual_key = %virtual_key,
"Streaming response (SSE) — wrapping with DLP SSE scanner"
);
let (parts, body) = response.into_parts();
let body = body
.collect()
.await
.map_err(|e| {
error!(error = %e, "Failed to read response body for DLP scan");
error_response(
StatusCode::INTERNAL_SERVER_ERROR,
"Failed to process response",
)
})?
.to_bytes();

let (redacted, redacted_names) = state.dlp_scanner.redact_all(&body);
let dlp_body =
crate::translate::wrap_body_with_dlp_sse_stream(body, state.dlp_scanner.clone());
Response::from_parts(parts, dlp_body)
} else {
trace!("Streaming response, DLP response scanning disabled");
response
}
} else {
let (parts, body) = response.into_parts();
let body_bytes = body
.collect()
.await
.map_err(|e| {
error!(error = %e, "Failed to read response body");
error_response(
StatusCode::INTERNAL_SERVER_ERROR,
"Failed to process response",
)
})?
.to_bytes();

// Count tokens from the upstream `usage` block before any DLP mutation.
state.stats.record_tokens_from_usage(&body_bytes);

if state.dlp_scanner.scan_responses() {
let (redacted, redacted_names) = state.dlp_scanner.redact_all(&body_bytes);
if !redacted_names.is_empty() {
warn!(
method = %method,
Expand All @@ -669,23 +729,11 @@ async fn handle_request(
parts.headers.remove("content-length");
Response::from_parts(parts, Body::from(redacted_bytes))
} else {
Response::from_parts(parts, Body::from(body))
Response::from_parts(parts, Body::from(body_bytes))
}
} else {
debug!(
method = %method,
path = %path,
virtual_key = %virtual_key,
"Streaming response (SSE) — wrapping with DLP SSE scanner"
);
let (parts, body) = response.into_parts();
let dlp_body =
crate::translate::wrap_body_with_dlp_sse_stream(body, state.dlp_scanner.clone());
Response::from_parts(parts, dlp_body)
Response::from_parts(parts, Body::from(body_bytes))
}
} else {
trace!("Response DLP scanning disabled");
response
};

Ok(response)
Expand Down
19 changes: 19 additions & 0 deletions src/app/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ use crate::email::{
use crate::keys::{KeyManager, KeySource, ResolvedKey};
use crate::oauth::OAuthRegistry;
use crate::proxy::ProxyClient;
use crate::stats::Stats;

fn make_app(upstream_url: &str) -> axum::Router {
let mut key_map = BTreeMap::new();
Expand Down Expand Up @@ -75,6 +76,7 @@ fn make_app(upstream_url: &str) -> axum::Router {
email_policy: None,
email_accounts: Arc::new(BTreeMap::new()),
email_service: Arc::new(EmailService::mock_disabled()),
stats: Arc::new(Stats::new(None)),
};

build_router(state)
Expand Down Expand Up @@ -117,6 +119,7 @@ fn make_app_with_anthropic(upstream_url: &str) -> axum::Router {
email_policy: None,
email_accounts: Arc::new(BTreeMap::new()),
email_service: Arc::new(EmailService::mock_disabled()),
stats: Arc::new(Stats::new(None)),
};

build_router(state)
Expand Down Expand Up @@ -568,6 +571,8 @@ openai_base_url = "https://api.openai.com"
[[keys]]
virtual_key = "vk-1"
real_key = "sk-real-1"
[stats]
persist_path = "/etc/clawshell/stats.json"
"#;
let config = Config::parse(toml_str).unwrap();
let state = AppState::from_config(&config).unwrap();
Expand Down Expand Up @@ -597,6 +602,8 @@ provider = "openai"
virtual_key = "vk-ant"
real_key = "sk-ant-key"
provider = "anthropic"
[stats]
persist_path = "/etc/clawshell/stats.json"
"#;
let config = Config::parse(toml_str).unwrap();
let state = AppState::from_config(&config).unwrap();
Expand Down Expand Up @@ -635,6 +642,9 @@ email = "bot@gmail.com"
app_password = "abcd efgh ijkl mnop"
imap_host = "imap.gmail.com"
imap_port = 993

[stats]
persist_path = "/etc/clawshell/stats.json"
"#;
let config = Config::parse(toml_str).unwrap();
let state = AppState::from_config(&config).unwrap();
Expand Down Expand Up @@ -665,6 +675,9 @@ allow_senders = ["alice@example.com"]
virtual_key = "vk-email"
email = "bot@gmail.com"
app_password = "abcd efgh ijkl mnop"

[stats]
persist_path = "/etc/clawshell/stats.json"
"#;
let config = Config::parse(toml_str).unwrap();
let state = AppState::from_config(&config).unwrap();
Expand Down Expand Up @@ -708,6 +721,7 @@ async fn test_proxy_error_on_unreachable_upstream() {
email_policy: None,
email_accounts: Arc::new(BTreeMap::new()),
email_service: Arc::new(EmailService::mock_disabled()),
stats: Arc::new(Stats::new(None)),
};

let app = build_router(state);
Expand Down Expand Up @@ -855,6 +869,7 @@ async fn test_anthropic_dlp_blocks_sensitive_data() {
email_policy: None,
email_accounts: Arc::new(BTreeMap::new()),
email_service: Arc::new(EmailService::mock_disabled()),
stats: Arc::new(Stats::new(None)),
};
let app = build_router(state);

Expand Down Expand Up @@ -957,6 +972,7 @@ async fn test_openai_and_openrouter_keys_map_to_distinct_real_keys() {
email_policy: None,
email_accounts: Arc::new(BTreeMap::new()),
email_service: Arc::new(EmailService::mock_disabled()),
stats: Arc::new(Stats::new(None)),
});

let openai_req = Request::builder()
Expand Down Expand Up @@ -1028,6 +1044,7 @@ fn make_app_with_redact(upstream_url: &str) -> axum::Router {
email_policy: None,
email_accounts: Arc::new(BTreeMap::new()),
email_service: Arc::new(EmailService::mock_disabled()),
stats: Arc::new(Stats::new(None)),
};

build_router(state)
Expand Down Expand Up @@ -1235,6 +1252,7 @@ async fn test_response_dlp_disabled() {
email_policy: None,
email_accounts: Arc::new(BTreeMap::new()),
email_service: Arc::new(EmailService::mock_disabled()),
stats: Arc::new(Stats::new(None)),
};
let app = build_router(state);

Expand Down Expand Up @@ -1598,6 +1616,7 @@ fn make_email_app(
email_policy: Some(policy),
email_accounts: Arc::new(email_accounts),
email_service,
stats: Arc::new(Stats::new(None)),
};

build_router(state)
Expand Down
Loading
Loading