Add incoming webhooks, fix stale service worker and push re-registration

- Webhooks for automations: per-job secret, Gitea/GitHub HMAC signatures,
  payload appended to the instructions; settings on the automations page
- Service worker gets a new file name per build (proxies that cache *.js
  served a stale worker, which broke updates and Android installation);
  /sw.js is a kill switch that removes old workers
- Push: devices re-register their subscription with the server on start and
  before a test; the test reports why a push service rejected a message

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Richard ZinkandClaude Opus 5.5 committed 2026-11-09 00:38:04 +01:00
1 parent 6dffcb501a
commit 10d4b64289
16 files changed
+267 -18

No files matched your search

+57 -2
View File
@@ -137,6 +137,7 @@ pub fn router(app: AppState) -> Router {
.route("/jobs", get(list_jobs).post(register_job))
.route("/jobs/{id}", axum::routing::patch(update_job).delete(delete_job))
.route("/jobs/{id}/run", post(run_job))
.route("/jobs/{id}/webhook", post(job_webhook_settings))
.route("/terminals", get(list_terminals).post(open_terminal))
.route("/terminals/{id}", delete(close_terminal))
.route("/terminals/{id}/ws", get(terminal_ws))
@@ -160,6 +161,7 @@ pub fn router(app: AppState) -> Router {
Router::new()
.nest("/api", api)
.route("/hooks/{id}", post(incoming_webhook))
.route("/cli/install.sh", get(cli_install_script))
.route("/cli/download/{target}", get(cli_download))
.fallback_service(static_files)
@@ -1254,6 +1256,59 @@ async fn delete_job(Extension(user): Extension<User>, State(app): State<AppState
Ok(Json(json!({"ok": true})))
}
#[derive(Deserialize)]
struct WebhookSettings {
enable: bool,
}
/// Enable (new secret, shown once) or disable the webhook of an automation.
async fn job_webhook_settings(Extension(user): Extension<User>, State(app): State<AppState>, headers: axum::http::HeaderMap, Path(id): Path<String>, Json(b): Json<WebhookSettings>) -> ApiResult {
let job = own_job(&app, &user, &id)?;
if !b.enable {
app.store.job_set_webhook(&job.id, None)?;
app.audit(Some(&user), "job.webhook.off", &job.path, "");
return Ok(Json(json!({"ok": true})));
}
let secret = crate::auth::random_token(24);
app.store.job_set_webhook(&job.id, Some(&secret))?;
app.audit(Some(&user), "job.webhook.on", &job.path, "");
let base = public_base(&app, &headers);
Ok(Json(json!({"url": format!("{base}/hooks/{}", job.id), "secret": secret})))
}
#[derive(Deserialize)]
struct HookQuery {
secret: Option<String>,
}
/// Public webhook endpoint: starts the automation with the request body as input.
async fn incoming_webhook(State(app): State<AppState>, Path(id): Path<String>, Query(q): Query<HookQuery>, headers: axum::http::HeaderMap, body: bytes::Bytes) -> Response {
let ip = crate::auth::client_ip(&headers);
let Ok(Some(job)) = app.store.job(&id) else { return (StatusCode::NOT_FOUND, "unknown webhook").into_response() };
let Ok(Some(secret)) = app.store.job_webhook_secret(&id) else { return (StatusCode::NOT_FOUND, "unknown webhook").into_response() };
if !crate::jobs::verify_webhook(&secret, &headers, q.secret.as_deref(), &body) {
app.store.audit(None, "", "webhook.denied", &id, &ip);
return (StatusCode::UNAUTHORIZED, "invalid secret or signature").into_response();
}
if !job.enabled {
return (StatusCode::CONFLICT, "the automation is disabled").into_response();
}
if job.last_session_id.as_deref().is_some_and(|s| app.is_busy(s)) {
return (StatusCode::CONFLICT, "the automation is still running").into_response();
}
if job.last_run.is_some_and(|t| now_ms() - t < 10_000) {
return (StatusCode::TOO_MANY_REQUESTS, "too many calls – wait a few seconds").into_response();
}
let payload = crate::jobs::payload_text(&headers, &body);
match crate::jobs::run_with(&app, &job, "webhook", Some(payload)).await {
Ok(sid) => {
app.store.audit(None, "", "webhook.run", &id, &ip);
(StatusCode::ACCEPTED, Json(json!({"ok": true, "session_id": sid}))).into_response()
}
Err(e) => (StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),
}
}
async fn run_job(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
let job = own_job(&app, &user, &id)?;
let sid = crate::jobs::run(&app, &job, "manual").await.map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
@@ -1348,8 +1403,8 @@ async fn push_unsubscribe(Extension(user): Extension<User>, State(app): State<Ap
}
async fn push_test(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
let sent = crate::push::send_to_user(&app, &user.id, &json!({"kind": "test", "title": "CodeCraft", "body": "Notifications work.", "tag": "test", "url": "/"})).await;
Ok(Json(json!({"sent": sent})))
let (sent, errors) = crate::push::send_to_user(&app, &user.id, &json!({"kind": "test", "title": "CodeCraft", "body": "Notifications work.", "tag": "test", "url": "/"})).await;
Ok(Json(json!({"sent": sent, "errors": errors})))
}
async fn ws(Extension(user): Extension<User>, State(app): State<AppState>, upgrade: WebSocketUpgrade) -> Response {
+75
View File
@@ -38,6 +38,8 @@ pub struct Job {
pub last_session_id: Option<String>,
pub last_status: Option<String>,
pub created_at: i64,
/// Webhook enabled (the secret – also the HMAC key – is only shown once when created).
pub webhook: bool,
}
/// Parsed job file.
@@ -117,6 +119,11 @@ pub fn describe(job: &Job) -> Value {
/// Start a run now. Returns the new session id.
pub async fn run(app: &Arc<App>, job: &Job, trigger: &str) -> Result<String> {
run_with(app, job, trigger, None).await
}
/// Start a run; `payload` (webhook body) is appended to the instructions.
pub async fn run_with(app: &Arc<App>, job: &Job, trigger: &str, payload: Option<String>) -> Result<String> {
let path = PathBuf::from(&job.path);
let spec = match parse(&path) {
Ok(s) => s,
@@ -161,6 +168,10 @@ pub async fn run(app: &Arc<App>, job: &Job, trigger: &str) -> Result<String> {
"<automation name=\"{}\" trigger=\"{trigger}\" file=\"{}\">\nThis is an unattended, scheduled run – nobody is watching live. Work autonomously, avoid questions unless you are truly blocked, and end with a short report of what you did and found.\n</automation>\n\n{}",
spec.name, job.path, spec.prompt
);
let text = match payload {
Some(p) => format!("{text}\n\n<webhook-payload>\n{p}\n</webhook-payload>\nThe automation was triggered by this webhook call – use the payload as input."),
None => text,
};
crate::runner::submit(app.clone(), session.id.clone(), vec![Part::Text { text }], "user")?;
// Watch for the end (or the timeout).
@@ -375,3 +386,67 @@ mod tests {
std::fs::remove_dir_all(&dir).unwrap();
}
}
// --- incoming webhooks -------------------------------------------------------------
fn constant_eq(a: &[u8], b: &[u8]) -> bool {
a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
}
fn hmac_hex(secret: &str, body: &[u8]) -> String {
use hmac::{KeyInit, Mac};
let mut mac = <hmac::Hmac<sha2::Sha256> as KeyInit>::new_from_slice(secret.as_bytes()).expect("hmac key");
mac.update(body);
mac.finalize().into_bytes().iter().map(|b| format!("{b:02x}")).collect()
}
/// Check a webhook call: shared secret (header/query) or Gitea/GitHub HMAC signature.
pub fn verify_webhook(secret: &str, headers: &axum::http::HeaderMap, query_secret: Option<&str>, body: &[u8]) -> bool {
let h = |k: &str| headers.get(k).and_then(|v| v.to_str().ok()).map(|v| v.trim().to_string());
if let Some(given) = h("x-codecraft-secret").or(query_secret.map(String::from)) {
return constant_eq(given.as_bytes(), secret.as_bytes());
}
let expected = hmac_hex(secret, body);
if let Some(sig) = h("x-gitea-signature").or(h("x-gogs-signature")) {
return constant_eq(sig.to_lowercase().as_bytes(), expected.as_bytes());
}
if let Some(sig) = h("x-hub-signature-256") {
return constant_eq(sig.trim_start_matches("sha256=").to_lowercase().as_bytes(), expected.as_bytes());
}
false
}
/// The payload as text for the prompt (pretty JSON, event name, size-limited).
pub fn payload_text(headers: &axum::http::HeaderMap, body: &[u8]) -> String {
let event = ["x-gitea-event", "x-github-event", "x-gitlab-event", "x-event-key"]
.iter()
.find_map(|k| headers.get(*k).and_then(|v| v.to_str().ok()))
.map(|e| format!("event: {e}\n"))
.unwrap_or_default();
let text = match serde_json::from_slice::<Value>(body) {
Ok(v) => serde_json::to_string_pretty(&v).unwrap_or_default(),
Err(_) => String::from_utf8_lossy(body).into_owned(),
};
let mut text: String = text.chars().take(20_000).collect();
if body.len() > 20_000 {
text.push_str("\n… (payload truncated)");
}
format!("{event}{text}")
}
#[cfg(test)]
mod webhook_tests {
#[test]
fn signatures() {
let mut h = axum::http::HeaderMap::new();
let sig = super::hmac_hex("s3cret", b"{\"a\":1}");
h.insert("x-hub-signature-256", format!("sha256={sig}").parse().unwrap());
assert!(super::verify_webhook("s3cret", &h, None, b"{\"a\":1}"));
assert!(!super::verify_webhook("other", &h, None, b"{\"a\":1}"));
let mut h2 = axum::http::HeaderMap::new();
h2.insert("x-gitea-signature", sig.parse().unwrap());
assert!(super::verify_webhook("s3cret", &h2, None, b"{\"a\":1}"));
assert!(super::verify_webhook("abc", &axum::http::HeaderMap::new(), Some("abc"), b""));
assert!(!super::verify_webhook("abc", &axum::http::HeaderMap::new(), None, b""));
}
}
+6 -4
View File
@@ -138,23 +138,25 @@ fn vapid_header(key: &SecretKey, endpoint: &str, subject: &str) -> Result<String
Ok(format!("vapid t={jwt}, k={k}"))
}
/// Send a notification to all devices of a user.
pub async fn send_to_user(app: &App, user_id: &str, payload: &Value) -> usize {
/// Send a notification to all devices of a user: (delivered, error messages).
pub async fn send_to_user(app: &App, user_id: &str, payload: &Value) -> (usize, Vec<String>) {
let subs = app.store.push_subscriptions(user_id).unwrap_or_default();
let mut sent = 0;
let mut errors = Vec::new();
for sub in subs {
match send_one(app, &sub.endpoint, &sub.p256dh, &sub.auth, payload).await {
Ok(()) => sent += 1,
Err(e) => {
let gone = e.to_string().contains("410") || e.to_string().contains("404");
tracing::debug!("push to {}: {e:#}", sub.endpoint);
tracing::warn!("push to {}…: {e:#}", sub.endpoint.chars().take(40).collect::<String>());
errors.push(format!("{e:#}").chars().take(200).collect());
if gone {
let _ = app.store.push_remove(&sub.endpoint);
}
}
}
}
sent
(sent, errors)
}
async fn send_one(app: &App, endpoint: &str, p256dh: &str, auth: &str, payload: &Value) -> Result<()> {
+15 -3
View File
@@ -161,6 +161,7 @@ impl Store {
("sessions", "owner_id", "ALTER TABLE sessions ADD COLUMN owner_id TEXT"),
("usage", "user_id", "ALTER TABLE usage ADD COLUMN user_id TEXT"),
("users", "monthly_quota", "ALTER TABLE users ADD COLUMN monthly_quota REAL"),
("jobs", "webhook_secret", "ALTER TABLE jobs ADD COLUMN webhook_secret TEXT"),
] {
let exists: bool = conn
.prepare(&format!("SELECT 1 FROM pragma_table_info('{table}') WHERE name = ?1"))?
@@ -975,13 +976,14 @@ impl Store {
last_session_id: r.get(5)?,
last_status: r.get(6)?,
created_at: r.get(7)?,
webhook: r.get::<_, Option<String>>(8)?.is_some(),
})
}
pub fn jobs(&self, owner: Option<&str>) -> Result<Vec<crate::jobs::Job>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, owner_id, path, enabled, last_run, last_session_id, last_status, created_at FROM jobs
"SELECT id, owner_id, path, enabled, last_run, last_session_id, last_status, created_at, webhook_secret FROM jobs
WHERE ?1 IS NULL OR owner_id = ?1 ORDER BY created_at",
)?;
let rows = stmt.query_map([owner], Self::job_row)?;
@@ -991,7 +993,7 @@ impl Store {
pub fn job(&self, id: &str) -> Result<Option<crate::jobs::Job>> {
let conn = self.conn.lock().unwrap();
Ok(conn
.query_row("SELECT id, owner_id, path, enabled, last_run, last_session_id, last_status, created_at FROM jobs WHERE id = ?1", [id], Self::job_row)
.query_row("SELECT id, owner_id, path, enabled, last_run, last_session_id, last_status, created_at, webhook_secret FROM jobs WHERE id = ?1", [id], Self::job_row)
.optional()?)
}
@@ -1007,7 +1009,7 @@ impl Store {
}
let conn = self.conn.lock().unwrap();
Ok(conn.query_row(
"SELECT id, owner_id, path, enabled, last_run, last_session_id, last_status, created_at FROM jobs WHERE path = ?1",
"SELECT id, owner_id, path, enabled, last_run, last_session_id, last_status, created_at, webhook_secret FROM jobs WHERE path = ?1",
[path],
Self::job_row,
)?)
@@ -1039,6 +1041,16 @@ impl Store {
Ok(())
}
/// Set (Some) or remove (None) the webhook secret of a job.
pub fn job_set_webhook(&self, id: &str, secret: Option<&str>) -> Result<()> {
self.conn.lock().unwrap().execute("UPDATE jobs SET webhook_secret = ?2 WHERE id = ?1", rusqlite::params![id, secret])?;
Ok(())
}
pub fn job_webhook_secret(&self, id: &str) -> Result<Option<String>> {
Ok(self.conn.lock().unwrap().query_row("SELECT webhook_secret FROM jobs WHERE id = ?1", [id], |r| r.get(0)).optional()?.flatten())
}
pub fn job_delete(&self, id: &str) -> Result<()> {
self.conn.lock().unwrap().execute("DELETE FROM jobs WHERE id = ?1", [id])?;
Ok(())