tatara_github_watcher/
handler.rs

1//! axum HTTP handler — verify signature, dispatch on event kind, apply
2//! resulting Allocation via kube-rs.
3
4use std::sync::Arc;
5
6use axum::body::Bytes;
7use axum::extract::State;
8use axum::http::{HeaderMap, StatusCode};
9use axum::response::IntoResponse;
10use kube::api::{Api, DeleteParams, PostParams};
11use kube::Client;
12use tracing::{info, warn};
13
14use tatara_process::allocation::EphemeralAllocation;
15
16use crate::allocation_factory::{allocation_name, build_allocation, FactoryError};
17use crate::config::WatcherConfig;
18use crate::event::{EventKind, PullRequestEvent};
19use crate::verify::verify_signature;
20
21/// Handler state shared across requests.
22#[derive(Clone)]
23pub struct HandlerState {
24    pub config: Arc<WatcherConfig>,
25    pub kube: Client,
26}
27
28/// POST handler for GitHub webhooks.
29pub async fn webhook(
30    State(state): State<HandlerState>,
31    headers: HeaderMap,
32    body: Bytes,
33) -> impl IntoResponse {
34    // 1. Verify HMAC.
35    let sig_header = headers
36        .get("X-Hub-Signature-256")
37        .and_then(|v| v.to_str().ok())
38        .unwrap_or("");
39    if let Err(e) = verify_signature(sig_header, &body, state.config.secret.as_bytes()) {
40        warn!(error = %e, "webhook signature verification failed");
41        return (StatusCode::UNAUTHORIZED, format!("signature: {e}")).into_response();
42    }
43
44    // 2. Dispatch on event kind.
45    let event_header = headers
46        .get("X-GitHub-Event")
47        .and_then(|v| v.to_str().ok())
48        .unwrap_or("");
49    let kind = EventKind::from_header(event_header);
50
51    match kind {
52        EventKind::PullRequest => handle_pr_event(&state, &body).await,
53        EventKind::Push => {
54            // Push events handled by a separate path (e.g., main-branch
55            // attestation runs). v0 just acknowledges.
56            (StatusCode::OK, "push event acknowledged (not allocated)").into_response()
57        }
58        EventKind::Other => (StatusCode::OK, "event ignored").into_response(),
59    }
60}
61
62async fn handle_pr_event(state: &HandlerState, body: &[u8]) -> axum::response::Response {
63    let evt: PullRequestEvent = match serde_json::from_slice(body) {
64        Ok(e) => e,
65        Err(e) => {
66            warn!(error = %e, "failed to parse PR event");
67            return (StatusCode::BAD_REQUEST, format!("parse: {e}")).into_response();
68        }
69    };
70
71    // Repo allowlist.
72    if !state.config.allow_repos.is_empty() && !repo_allowed(&evt.repository.full_name, &state.config.allow_repos) {
73        info!(repo = %evt.repository.full_name, "repo not in allowlist; skipping");
74        return (StatusCode::OK, "repo not in allowlist").into_response();
75    }
76
77    use crate::event::PrAction;
78    match evt.action {
79        PrAction::Closed => {
80            // Delete the allocation; pool reconciler returns the member.
81            let name = allocation_name(&evt.repository.full_name, evt.number);
82            let api: Api<EphemeralAllocation> =
83                Api::namespaced(state.kube.clone(), &state.config.namespace);
84            match api.delete(&name, &DeleteParams::default()).await {
85                Ok(_) => {
86                    info!(
87                        namespace = %state.config.namespace,
88                        allocation = %name,
89                        "closed PR → deleted Allocation"
90                    );
91                    (StatusCode::OK, "allocation deleted").into_response()
92                }
93                Err(kube::Error::Api(e)) if e.code == 404 => {
94                    (StatusCode::OK, "allocation already gone").into_response()
95                }
96                Err(e) => {
97                    warn!(error = %e, "delete failed");
98                    (StatusCode::INTERNAL_SERVER_ERROR, format!("delete: {e}"))
99                        .into_response()
100                }
101            }
102        }
103        PrAction::Opened | PrAction::Reopened | PrAction::Synchronize => {
104            // Build + create-or-replace the allocation.
105            let alloc = match build_allocation(
106                &evt,
107                &state.config.namespace,
108                state.config.pin_pool.as_deref(),
109                state.config.include_drafts,
110            ) {
111                Ok(a) => a,
112                Err(FactoryError::DraftExcluded) => {
113                    info!("draft PR — skipping allocation");
114                    return (StatusCode::OK, "draft excluded").into_response();
115                }
116                Err(FactoryError::NotAllocatable(_)) => {
117                    return (StatusCode::OK, "action not allocatable").into_response();
118                }
119            };
120            let api: Api<EphemeralAllocation> =
121                Api::namespaced(state.kube.clone(), &state.config.namespace);
122            match api.create(&PostParams::default(), &alloc).await {
123                Ok(_) => {
124                    info!(
125                        namespace = %state.config.namespace,
126                        allocation = alloc.metadata.name.as_deref().unwrap_or("?"),
127                        pr_number = evt.number,
128                        repo = %evt.repository.full_name,
129                        "PR event → created Allocation"
130                    );
131                    (StatusCode::CREATED, "allocation created").into_response()
132                }
133                Err(kube::Error::Api(e)) if e.code == 409 => {
134                    // Already exists — refresh via PATCH (synchronize event).
135                    (StatusCode::OK, "allocation already exists (synchronize)").into_response()
136                }
137                Err(e) => {
138                    warn!(error = %e, "create allocation failed");
139                    (StatusCode::INTERNAL_SERVER_ERROR, format!("create: {e}"))
140                        .into_response()
141                }
142            }
143        }
144        PrAction::Other => (StatusCode::OK, "action ignored").into_response(),
145    }
146}
147
148fn repo_allowed(repo: &str, allowlist: &[String]) -> bool {
149    allowlist.iter().any(|p| repo_matches(p, repo))
150}
151
152fn repo_matches(pattern: &str, repo: &str) -> bool {
153    if let Some(prefix) = pattern.strip_suffix("/*") {
154        repo.starts_with(&format!("{prefix}/"))
155    } else {
156        pattern == repo
157    }
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163
164    #[test]
165    fn repo_matches_exact() {
166        assert!(repo_matches("pleme-io/akeyless", "pleme-io/akeyless"));
167        assert!(!repo_matches("pleme-io/akeyless", "pleme-io/other"));
168    }
169
170    #[test]
171    fn repo_matches_org_wildcard() {
172        assert!(repo_matches("pleme-io/*", "pleme-io/akeyless"));
173        assert!(repo_matches("pleme-io/*", "pleme-io/tatara"));
174        assert!(!repo_matches("pleme-io/*", "drzln/dotfiles"));
175    }
176
177    #[test]
178    fn empty_allowlist_skipped_at_caller() {
179        // The caller's check `!allowlist.is_empty()` gates this function;
180        // sanity test that an empty allowlist would reject everything if
181        // called directly.
182        assert!(!repo_allowed("anything", &[]));
183    }
184
185    #[test]
186    fn allowlist_with_one_pattern_filters() {
187        let allow = vec!["pleme-io/*".to_string()];
188        assert!(repo_allowed("pleme-io/akeyless", &allow));
189        assert!(!repo_allowed("drzln/dotfiles", &allow));
190    }
191}
tatara_github_watcher/handler.rs

tatara_github_watcher/
handler.rs
