diff --git a/.github/workflows/terraform-google-maple.yml b/.github/workflows/terraform-google-maple.yml new file mode 100644 index 0000000000..941e955344 --- /dev/null +++ b/.github/workflows/terraform-google-maple.yml @@ -0,0 +1,37 @@ +name: Terraform module (Google Cloud) + +on: + push: + branches: [main] + paths: + - "deploy/terraform-google-maple/**" + - ".github/workflows/terraform-google-maple.yml" + pull_request: + paths: + - "deploy/terraform-google-maple/**" + - ".github/workflows/terraform-google-maple.yml" + workflow_dispatch: + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + test: + name: Format, validate and plan against mocked providers + runs-on: ubuntu-latest + defaults: + run: + working-directory: deploy/terraform-google-maple + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v6 + - uses: opentofu/setup-opentofu@a1320f892987e89d278cc92dc5adc984fb93aca4 # v2.0.2 + with: + tofu_version: 1.13.1 + - run: tofu fmt -recursive -check -diff + - run: tofu init -backend=false + - run: tofu validate + - run: tofu test diff --git a/apps/api/src/routes/v2/integrations-gcp.http.test.ts b/apps/api/src/routes/v2/integrations-gcp.http.test.ts index afb28127b8..63c396a8aa 100644 --- a/apps/api/src/routes/v2/integrations-gcp.http.test.ts +++ b/apps/api/src/routes/v2/integrations-gcp.http.test.ts @@ -216,11 +216,26 @@ describe("v2 gcp integration over HTTP", () => { application_logs: ["cloud_run", "cloud_functions", "app_engine", "gke"], }) expect(both.status).toBe(200) - expect(Object.keys(both.body).sort()).toEqual(["cleanup_script", "object", "setup_script"]) + expect(Object.keys(both.body).sort()).toEqual([ + "cleanup_script", + "object", + "setup_script", + "terraform", + ]) expect(both.body.object).toBe("gcp_connector.setup_scripts") + // The subscription is given no secret; only the script's own reports carry it. expect(both.body.setup_script).toMatch( - /PUSH_ENDPOINT='https:\/\/ingest\.test\/v1\/logpush\/gcp\/[0-9a-f-]{36}\?secret=maple_gcp_/, + /PUSH_ENDPOINT='https:\/\/ingest\.test\/v1\/logpush\/gcp\/[0-9a-f-]{36}'\n/, ) + expect(both.body.setup_script).toMatch( + /REPORT_ENDPOINT='https:\/\/ingest\.test\/v1\/logpush\/gcp\/[0-9a-f-]{36}\?secret=maple_gcp_/, + ) + expect(both.body.terraform).toMatch(/^ {2}scope_type += "folder"$/m) + expect(both.body.terraform).toMatch(/^ {2}project_id += "acme-host"$/m) + expect(both.body.terraform).toContain( + 'NOT protoPayload.methodName=\\"io.k8s.coordination.v1.leases.update\\"', + ) + expect(both.body.terraform).not.toMatch(/maple_gcp_[A-Za-z0-9_-]{32}|secret=/) expect(both.body.setup_script).toContain("PROJECT_ID='acme-host'") expect(both.body.setup_script).toContain("SCOPE_ID='123456789012'") expect(both.body.setup_script).toContain('--folder="$SCOPE_ID" --include-children') @@ -262,7 +277,7 @@ describe("v2 gcp integration over HTTP", () => { expect(reader.body.setup_script).toContain( "MAPLE_SERVICE_ACCOUNT='collector@maple-prod.iam.gserviceaccount.com'", ) - expect(reader.body.setup_script).not.toContain("service-accounts delete") + expect(reader.body.setup_script).not.toContain('service-accounts delete "$SERVICE_ACCOUNT_EMAIL"') await harness.dispose() }) @@ -295,6 +310,11 @@ describe("v2 gcp integration over HTTP", () => { const turnOff = await harness.request("PATCH", path, admin.secret, { logs_enabled: false }) expect(turnOff.status).toBe(400) expect(turnOff.body.error.code).toBe("integration_request_invalid") + + // Without a Google identity there is no account to read as. + const check = await harness.request("POST", `${path}/check_access`, admin.secret) + expect(check.status).toBe(409) + expect(check.body.error.code).toBe("gcp_metrics_unavailable") await harness.dispose() }) @@ -310,6 +330,7 @@ describe("v2 gcp integration over HTTP", () => { ["POST", CONNECTORS, { scope_type: "project", scope_id: "other-project" }], ["PATCH", path, { metrics_enabled: true }], ["POST", `${path}/setup_scripts`, {}], + ["POST", `${path}/check_access`, undefined], ["DELETE", path, undefined], ] as const) { const asMember = await harness.request(method, target, member.secret, body) diff --git a/apps/api/src/routes/v2/integrations-gcp.http.ts b/apps/api/src/routes/v2/integrations-gcp.http.ts index dbfb87d6ec..374948a99a 100644 --- a/apps/api/src/routes/v2/integrations-gcp.http.ts +++ b/apps/api/src/routes/v2/integrations-gcp.http.ts @@ -1,5 +1,5 @@ import { HttpApiBuilder } from "effect/http-api" -import { CurrentTenant } from "@maple/domain/http" +import { CurrentTenant, GcpMetricsUnavailableError } from "@maple/domain/http" import type { V2GcpConnector } from "@maple/domain/http/v2" import { MapleApiV2, @@ -14,6 +14,7 @@ import { GcpConnectorService, type GcpConnector, } from "@maple/backend/services/integrations/GcpConnectorService" +import { GcpPromQlService } from "@maple/backend/services/integrations/GcpPromQlService" // Reading the status is open to every member. Everything else is admin-only: a connector decides // what may write logs into the org and what Maple reads, and its setup script carries the push @@ -51,6 +52,7 @@ const auditMetadata = (connector: GcpConnector) => ({ export const HttpV2GcpIntegrationsLive = HttpApiBuilder.group(MapleApiV2, "gcpIntegration", (handlers) => Effect.gen(function* () { const gcp = yield* GcpConnectorService + const gcpPromQl = yield* GcpPromQlService return handlers .handle("status", () => @@ -109,9 +111,24 @@ export const HttpV2GcpIntegrationsLive = HttpApiBuilder.group(MapleApiV2, "gcpIn object: "gcp_connector.setup_scripts" as const, setup_script: scripts.setupScript, cleanup_script: scripts.cleanupScript, + terraform: scripts.terraform, } }), ) + .handle("checkAccess", ({ params }) => + Effect.gen(function* () { + const tenant = yield* CurrentTenant.Context + yield* requireAdmin(tenant.roles, adminOnly) + const checked = yield* gcpPromQl + .checkAccess(tenant.orgId, params.id) + .pipe( + Effect.catchTag("@maple/http/errors/GcpPromQlUnavailableError", (error) => + Effect.fail(new GcpMetricsUnavailableError({ message: error.message })), + ), + ) + return { object: "gcp_connector.access_check" as const, ...checked } + }), + ) .handle("deleteConnector", ({ params }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context diff --git a/apps/api/src/routes/v2/v2-test-support.ts b/apps/api/src/routes/v2/v2-test-support.ts index 56033f7649..5c6a254f3b 100644 --- a/apps/api/src/routes/v2/v2-test-support.ts +++ b/apps/api/src/routes/v2/v2-test-support.ts @@ -45,6 +45,7 @@ import { HttpV2ChatIntegrationsLive } from "./integrations-chat.http" import { HttpV2GcpIntegrationsLive } from "./integrations-gcp.http" import { HttpV2PlanetScaleIntegrationsLive } from "./integrations.http" import { GcpConnectorService } from "@maple/backend/services/integrations/GcpConnectorService" +import { GcpPromQlService } from "@maple/backend/services/integrations/GcpPromQlService" import { HttpV2ErrorIssuesLive } from "./error-issues.http" import { HttpV2AnomaliesLive } from "./anomalies.http" import { HttpV2InvestigationsLive } from "./investigations.http" @@ -158,7 +159,9 @@ const v2GroupLayersExceptOnboardingChecklist = (chatWorkspace: Layer.Layer`, signed by Google, naming +//! that account (`email`) and the audience the subscription was given (`aud`). +//! Only someone who may act as the account can make Pub/Sub sign as it, so the +//! token stands in for a secret in the push URL. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use base64::Engine; +use ring::signature::{RsaPublicKeyComponents, RSA_PKCS1_2048_8192_SHA256}; +use serde::de::DeserializeOwned; +use serde::Deserialize; +use tokio::sync::RwLock; + +use crate::telemetry::HttpClient; + +const GOOGLE_CERTS_URL: &str = "https://www.googleapis.com/oauth2/v3/certs"; +const GOOGLE_ISSUERS: [&str; 2] = ["https://accounts.google.com", "accounts.google.com"]; +/// Tolerated difference between Google's clock and this host's. +const CLOCK_SKEW_SECS: u64 = 60; +/// Google publishes a signing key well before it uses it, and answers with a +/// `max-age` of several hours. An hour keeps rotation far inside that. +const KEYS_TTL: Duration = Duration::from_hours(1); +/// A token naming a key not in hand fetches the keys again at most this often, +/// so made-up key ids can't turn every push into a fetch. +const REFETCH_INTERVAL: Duration = Duration::from_mins(1); + +/// An RSA public key, as the big-endian bytes of its modulus and exponent. +#[derive(Clone, Debug)] +pub struct RsaKey { + pub n: Vec, + pub e: Vec, +} + +/// Google's signing keys by key id. +pub type Keys = HashMap; + +/// What a valid token says. +#[derive(Debug, PartialEq, Eq)] +pub struct VerifiedPush { + /// The service account the subscription signs as. + pub email: String, + pub audience: String, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum TokenError { + /// Signed with a key not among those given. + UnknownKey, + /// Never a token to accept: malformed, forged, expired, not Google's. + Invalid(&'static str), +} + +#[derive(Debug)] +pub enum PushAuthError { + /// The push is refused. + Invalid(&'static str), + /// Google's keys could not be fetched: the push is not to blame and is + /// worth delivering again. + KeysUnavailable(String), +} + +#[derive(Deserialize)] +struct Header { + alg: String, + kid: Option, +} + +#[derive(Deserialize)] +struct Claims { + iss: String, + aud: String, + exp: u64, + #[serde(default)] + iat: u64, + email: Option, + #[serde(default)] + email_verified: bool, +} + +#[derive(Deserialize)] +struct Jwk { + kid: String, + n: String, + e: String, +} + +#[derive(Deserialize)] +struct JwkSet { + keys: Vec, +} + +fn decode_json(segment: &str) -> Option { + serde_json::from_slice(&URL_SAFE_NO_PAD.decode(segment).ok()?).ok() +} + +/// Checks a token's RS256 signature against `keys`, and its claims at +/// `now_secs`. The algorithm is fixed: a token that names another is refused. +pub fn verify_with_keys( + token: &str, + keys: &Keys, + now_secs: u64, +) -> Result { + let invalid = TokenError::Invalid; + let mut segments = token.split('.'); + let (Some(header), Some(claims), Some(signature), None) = ( + segments.next(), + segments.next(), + segments.next(), + segments.next(), + ) else { + return Err(invalid("not a JWT")); + }; + let parsed: Header = decode_json(header).ok_or(invalid("malformed header"))?; + if parsed.alg != "RS256" { + return Err(invalid("unsupported algorithm")); + } + let key = parsed + .kid + .as_deref() + .and_then(|kid| keys.get(kid)) + .ok_or(TokenError::UnknownKey)?; + let signature = URL_SAFE_NO_PAD + .decode(signature) + .map_err(|_| invalid("malformed signature"))?; + let signed = &token[..header.len() + 1 + claims.len()]; + RsaPublicKeyComponents { + n: &key.n, + e: &key.e, + } + .verify(&RSA_PKCS1_2048_8192_SHA256, signed.as_bytes(), &signature) + .map_err(|_| invalid("bad signature"))?; + + let claims: Claims = decode_json(claims).ok_or(invalid("malformed claims"))?; + if !GOOGLE_ISSUERS.contains(&claims.iss.as_str()) { + return Err(invalid("not issued by Google")); + } + if claims.exp.saturating_add(CLOCK_SKEW_SECS) < now_secs { + return Err(invalid("expired")); + } + if claims.iat > now_secs.saturating_add(CLOCK_SKEW_SECS) { + return Err(invalid("issued in the future")); + } + let email = claims + .email + .filter(|_| claims.email_verified) + .ok_or(invalid("no verified email"))?; + Ok(VerifiedPush { + email, + audience: claims.aud, + }) +} + +fn parse_keys(body: &[u8]) -> Result { + let set: JwkSet = serde_json::from_slice(body).map_err(|error| error.to_string())?; + set.keys + .into_iter() + .map(|jwk| { + let n = URL_SAFE_NO_PAD + .decode(&jwk.n) + .map_err(|error| error.to_string())?; + let e = URL_SAFE_NO_PAD + .decode(&jwk.e) + .map_err(|error| error.to_string())?; + Ok((jwk.kid, RsaKey { n, e })) + }) + .collect() +} + +enum Source { + Google(HttpClient), + Fixed, +} + +struct Cached { + keys: Arc, + fetched_at: Option, +} + +/// Verifies push tokens against Google's published keys, fetched on first use +/// and kept for `KEYS_TTL`. +pub struct GooglePushAuth { + source: Source, + cached: RwLock, +} + +impl std::fmt::Debug for GooglePushAuth { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("GooglePushAuth") + .finish_non_exhaustive() + } +} + +impl GooglePushAuth { + pub fn new(client: impl Into) -> Self { + Self { + source: Source::Google(client.into()), + cached: RwLock::new(Cached { + keys: Arc::new(Keys::new()), + fetched_at: None, + }), + } + } + + /// Never fetches: for tests and hosts without network access. + pub fn with_keys(keys: Keys) -> Self { + Self { + source: Source::Fixed, + cached: RwLock::new(Cached { + keys: Arc::new(keys), + fetched_at: Some(Instant::now()), + }), + } + } + + pub async fn verify(&self, token: &str, now_secs: u64) -> Result { + let keys = self.keys(false).await?; + let verified = match verify_with_keys(token, &keys, now_secs) { + // Google may have started signing with a key published since. + Err(TokenError::UnknownKey) => { + verify_with_keys(token, &*self.keys(true).await?, now_secs) + } + other => other, + }; + verified.map_err(|error| match error { + TokenError::UnknownKey => PushAuthError::Invalid("unknown key"), + TokenError::Invalid(reason) => PushAuthError::Invalid(reason), + }) + } + + async fn keys(&self, unknown_key: bool) -> Result, PushAuthError> { + let due = |cached: &Cached| match cached.fetched_at { + None => true, + Some(at) => at.elapsed() > KEYS_TTL || (unknown_key && at.elapsed() > REFETCH_INTERVAL), + }; + { + let cached = self.cached.read().await; + if !due(&cached) { + return Ok(Arc::clone(&cached.keys)); + } + } + let Source::Google(client) = &self.source else { + return Ok(Arc::clone(&self.cached.read().await.keys)); + }; + let mut cached = self.cached.write().await; + // Another push may have fetched while this one waited for the lock. + if !due(&cached) { + return Ok(Arc::clone(&cached.keys)); + } + match fetch_keys(client).await { + Ok(keys) => { + cached.keys = Arc::new(keys); + cached.fetched_at = Some(Instant::now()); + Ok(Arc::clone(&cached.keys)) + } + // Keys still in hand keep verifying until a fetch succeeds. + Err(_) if !cached.keys.is_empty() => Ok(Arc::clone(&cached.keys)), + Err(error) => Err(PushAuthError::KeysUnavailable(error)), + } + } +} + +async fn fetch_keys(client: &HttpClient) -> Result { + let response = client + .get(GOOGLE_CERTS_URL) + .send() + .await + .map_err(|error| format!("fetching Google's signing keys failed: {error}"))?; + if !response.status().is_success() { + return Err(format!( + "fetching Google's signing keys failed: HTTP {}", + response.status() + )); + } + let body = response + .bytes() + .await + .map_err(|error| format!("reading Google's signing keys failed: {error}"))?; + parse_keys(&body) +} + +#[cfg(test)] +mod tests { + use super::*; + use ring::rand::SystemRandom; + use ring::signature::{RsaKeyPair, RSA_PKCS1_SHA256}; + use serde_json::json; + + const TEST_KEY: &[u8] = include_bytes!("testdata/gcp-push-test-key.pk8"); + const TEST_KEY_N: &str = include_str!("testdata/gcp-push-test-key.n"); + const NOW: u64 = 1_792_000_000; + + fn keys() -> Keys { + Keys::from([( + "key-1".to_owned(), + RsaKey { + n: URL_SAFE_NO_PAD.decode(TEST_KEY_N.trim()).unwrap(), + e: vec![1, 0, 1], + }, + )]) + } + + fn sign(header: &serde_json::Value, claims: &serde_json::Value) -> String { + let encode = |value: &serde_json::Value| URL_SAFE_NO_PAD.encode(value.to_string()); + let signed = format!("{}.{}", encode(header), encode(claims)); + let pair = RsaKeyPair::from_pkcs8(TEST_KEY).unwrap(); + let mut signature = vec![0; pair.public().modulus_len()]; + pair.sign( + &RSA_PKCS1_SHA256, + &SystemRandom::new(), + signed.as_bytes(), + &mut signature, + ) + .unwrap(); + format!("{signed}.{}", URL_SAFE_NO_PAD.encode(signature)) + } + + fn claims() -> serde_json::Value { + json!({ + "iss": "https://accounts.google.com", + "aud": "https://ingest.test/v1/logpush/gcp/c1", + "exp": NOW + 3600, + "iat": NOW, + "email": "maple-push-0a1b@acme.iam.gserviceaccount.com", + "email_verified": true, + "sub": "1234", + }) + } + + fn token(claims: &serde_json::Value) -> String { + sign( + &json!({ "alg": "RS256", "kid": "key-1", "typ": "JWT" }), + claims, + ) + } + + #[test] + fn accepts_a_token_google_signed_for_a_verified_account() { + assert_eq!( + verify_with_keys(&token(&claims()), &keys(), NOW), + Ok(VerifiedPush { + email: "maple-push-0a1b@acme.iam.gserviceaccount.com".to_owned(), + audience: "https://ingest.test/v1/logpush/gcp/c1".to_owned(), + }) + ); + } + + #[test] + fn refuses_tokens_that_are_not_google_signed_current_and_verified() { + let changed = |key: &str, value: serde_json::Value| { + let mut claims = claims(); + claims[key] = value; + verify_with_keys(&token(&claims), &keys(), NOW) + }; + assert_eq!( + changed("iss", json!("https://evil.test")), + Err(TokenError::Invalid("not issued by Google")) + ); + assert_eq!( + changed("exp", json!(NOW - 61)), + Err(TokenError::Invalid("expired")) + ); + assert_eq!( + changed("iat", json!(NOW + 61)), + Err(TokenError::Invalid("issued in the future")) + ); + assert_eq!( + changed("email_verified", json!(false)), + Err(TokenError::Invalid("no verified email")) + ); + // Within the tolerated clock difference. + assert!(changed("exp", json!(NOW - 59)).is_ok()); + + // Another account's claims under a genuine signature. + let valid = token(&claims()); + let segments: Vec<&str> = valid.split('.').collect(); + let mut forged = claims(); + forged["email"] = json!("someone-else@acme.iam.gserviceaccount.com"); + let swapped = format!( + "{}.{}.{}", + segments[0], + URL_SAFE_NO_PAD.encode(forged.to_string()), + segments[2] + ); + assert_eq!( + verify_with_keys(&swapped, &keys(), NOW), + Err(TokenError::Invalid("bad signature")) + ); + + let none = sign(&json!({ "alg": "none", "kid": "key-1" }), &claims()); + assert_eq!( + verify_with_keys(&none, &keys(), NOW), + Err(TokenError::Invalid("unsupported algorithm")) + ); + let other_key = sign(&json!({ "alg": "RS256", "kid": "key-2" }), &claims()); + assert_eq!( + verify_with_keys(&other_key, &keys(), NOW), + Err(TokenError::UnknownKey) + ); + assert_eq!( + verify_with_keys("a.b", &keys(), NOW), + Err(TokenError::Invalid("not a JWT")) + ); + } + + #[test] + fn reads_google_jwk_sets() { + let parsed = parse_keys( + json!({ "keys": [{ "kid": "abc", "n": TEST_KEY_N.trim(), "e": "AQAB", "kty": "RSA", "alg": "RS256", "use": "sig" }] }) + .to_string() + .as_bytes(), + ) + .unwrap(); + assert_eq!(parsed["abc"].e, vec![1, 0, 1]); + assert!(verify_with_keys( + &sign(&json!({ "alg": "RS256", "kid": "abc" }), &claims()), + &parsed, + NOW + ) + .is_ok()); + } + + #[tokio::test] + async fn fixed_keys_verify_without_fetching() { + let auth = GooglePushAuth::with_keys(keys()); + assert!(auth.verify(&token(&claims()), NOW).await.is_ok()); + let other_key = sign(&json!({ "alg": "RS256", "kid": "key-2" }), &claims()); + assert!(matches!( + auth.verify(&other_key, NOW).await, + Err(PushAuthError::Invalid("unknown key")) + )); + } +} diff --git a/apps/ingest/src/lib.rs b/apps/ingest/src/lib.rs index f39f4baf07..22c1e5e57c 100644 --- a/apps/ingest/src/lib.rs +++ b/apps/ingest/src/lib.rs @@ -2,6 +2,7 @@ pub mod ai_session; pub mod aws; pub mod clickhouse_insert_mappings; pub mod gcp_logging; +pub mod gcp_push_auth; pub mod metrics; pub mod otel; pub mod otlp_json; diff --git a/apps/ingest/src/main.rs b/apps/ingest/src/main.rs index 5a29d2a077..5c0a4b5fdd 100644 --- a/apps/ingest/src/main.rs +++ b/apps/ingest/src/main.rs @@ -47,6 +47,7 @@ use maple_ingest::ai_session; use maple_ingest::aws::CredentialsProvider as AwsCredentialsProvider; use maple_ingest::clickhouse_insert_mappings::SCHEMA_VERSION as CLICKHOUSE_SCHEMA_VERSION; use maple_ingest::gcp_logging; +use maple_ingest::gcp_push_auth::{GooglePushAuth, PushAuthError}; use maple_ingest::metrics; use maple_ingest::otel::{ accept_internal_span, auth_internal_span, build_resource, decode_internal_span, @@ -817,6 +818,8 @@ struct GcpConnectorResolver { recent_health_writes: Cache<(String, bool), ()>, /// Setup reports per connector in the current `GCP_SETUP_REPORT_WINDOW`. recent_setup_reports: Cache>, + /// Verifies the ID tokens Pub/Sub signs pushes with. + push_auth: GooglePushAuth, } struct OrgRoutingResolver { @@ -900,6 +903,12 @@ trait KeyStore: Send + Sync { secret_hash: &str, ) -> Result, String>; + /// For a push signed by Google: the caller checks the signing account. + async fn fetch_gcp_connector_by_id( + &self, + connector_id: &str, + ) -> Result, String>; + async fn fetch_sampling_policy( &self, org_id: &str, @@ -962,7 +971,11 @@ struct ConnectorRow { #[derive(Clone, Debug)] struct GcpConnectorRow { org_id: String, + /// The host project: its push service account signs the connector's pushes. + project_id: String, logs_enabled: bool, + /// What a setup run last reported; `None` before any report. + applied_logs_enabled: Option, self_managed: bool, clickhouse_ready: bool, } @@ -1031,12 +1044,14 @@ struct GcpConnectorIdentity { org_id: String, secret_key_id: String, logs_enabled: bool, + confirms_log_forwarding: bool, } impl GcpConnectorIdentity { fn into_resolved(self, routing: &OrgRouting) -> ResolvedGcpConnector { ResolvedGcpConnector { logs_enabled: self.logs_enabled, + confirms_log_forwarding: self.confirms_log_forwarding, key: ResolvedIngestKey { org_id: self.org_id, key_type: IngestKeyType::Connector, @@ -1263,6 +1278,10 @@ struct ResolvedCloudflareConnector { struct ResolvedGcpConnector { /// False once the org turns log forwarding off; the sink may still push. logs_enabled: bool, + /// A signed push to a connector no setup run has reported on: one set up + /// with Terraform, which sends no report. The push itself proves the sink, + /// topic and subscription are in place. + confirms_log_forwarding: bool, key: ResolvedIngestKey, } @@ -2566,6 +2585,7 @@ async fn main() { config.lookup_hmac_key.clone(), org_routing_resolver, Duration::from_secs(config.ingest_key_cache_ttl_secs), + GooglePushAuth::new(http_client.clone()), ), telemetry_pipeline, http_client, @@ -4831,6 +4851,7 @@ async fn handle_gcp_logpush( State(state): State>, Path(connector_id): Path, Query(query): Query, + headers: HeaderMap, body: Bytes, ) -> Response { let start = Instant::now(); @@ -4854,6 +4875,7 @@ async fn handle_gcp_logpush( "maple.org_id" = tracing::field::Empty, "maple.gcp.connector_id" = tracing::field::Empty, "maple.gcp.setup_report" = tracing::field::Empty, + "maple.gcp.push_auth" = tracing::field::Empty, "maple.ingest.self_managed" = tracing::field::Empty, "maple.ingest.clickhouse_ready" = tracing::field::Empty, "maple.ingest.destination" = tracing::field::Empty, @@ -4861,7 +4883,18 @@ async fn handle_gcp_logpush( ); let span_handle = span.clone(); - let result = handle_gcp_logpush_inner(&state, &connector_id, query.secret.as_deref(), body) + // Set up by the current script or Terraform: a token Pub/Sub signed. + // Earlier subscriptions carry the secret in the URL. + let token = headers + .get(AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")); + let credential = match (token, query.secret.as_deref()) { + (Some(token), _) => Some(GcpPushCredential::Token(token)), + (None, Some(secret)) => Some(GcpPushCredential::Secret(secret)), + (None, None) => None, + }; + let result = handle_gcp_logpush_inner(&state, &connector_id, credential, body) .instrument(span) .await; let duration = start.elapsed().as_secs_f64(); @@ -4923,12 +4956,17 @@ async fn handle_gcp_logpush( } } +enum GcpPushCredential<'a> { + Token(&'a str), + Secret(&'a str), +} + /// Returns Ok((ack response, what the ack means)) or /// Err((ApiError, error_kind_label)). async fn handle_gcp_logpush_inner( state: &AppState, connector_id: &str, - secret: Option<&str>, + credential: Option>, body: Bytes, ) -> Result<(Response, GcpPushOutcome), (ApiError, &'static str)> { let unauthorized = || { @@ -4937,17 +4975,34 @@ async fn handle_gcp_logpush_inner( "auth", ) }; - let secret = secret.map(str::trim).filter(|value| !value.is_empty()); - let (Some(secret), true) = (secret, is_connector_id(connector_id)) else { - warn!("Missing GCP connector secret or malformed connector id"); + let credential = credential.filter(|credential| match credential { + GcpPushCredential::Token(value) | GcpPushCredential::Secret(value) => { + !value.trim().is_empty() + } + }); + let (Some(credential), true) = (credential, is_connector_id(connector_id)) else { + warn!("Missing GCP connector credentials or malformed connector id"); return Err(unauthorized()); }; Span::current().record("maple.gcp.connector_id", connector_id); - let resolved = state - .gcp_resolver - .resolve_connector(connector_id, secret) - .await + let lookup = match credential { + GcpPushCredential::Token(token) => { + Span::current().record("maple.gcp.push_auth", "oidc"); + state + .gcp_resolver + .resolve_connector_by_token(connector_id, token.trim()) + .await + } + GcpPushCredential::Secret(secret) => { + Span::current().record("maple.gcp.push_auth", "secret"); + state + .gcp_resolver + .resolve_connector(connector_id, secret.trim()) + .await + } + }; + let resolved = lookup .map_err(|error| { error!(error = %error, connector_id, "GCP connector resolution failed"); ( @@ -4962,6 +5017,12 @@ async fn handle_gcp_logpush_inner( let org_id = resolved.key.org_id.as_str(); Span::current().record("maple.org_id", org_id); + if resolved.confirms_log_forwarding { + state + .gcp_resolver + .confirm_log_forwarding(connector_id) + .await; + } Span::current().record("maple.ingest.self_managed", resolved.key.self_managed); Span::current().record( "maple.ingest.clickhouse_ready", @@ -5104,6 +5165,17 @@ async fn handle_gcp_logpush_inner( /// Connector ids are generated by the API; anything outside this alphabet is /// refused before it reaches this handler's span, its logs, or Postgres. +/// The push service account a connector's setup creates in its host project, +/// named as `gcpConnectorResourceNames` in `@maple/domain/gcp` names it. +fn gcp_push_service_account(connector_id: &str, project_id: &str) -> String { + let hex: String = connector_id + .chars() + .filter(|c| *c != '-') + .take(19) + .collect(); + format!("maple-push-{hex}@{project_id}.iam.gserviceaccount.com") +} + fn is_connector_id(value: &str) -> bool { (1..=128).contains(&value.len()) && value @@ -6129,6 +6201,7 @@ impl GcpConnectorResolver { lookup_hmac_key: String, routing: Arc, cache_ttl: Duration, + push_auth: GooglePushAuth, ) -> Self { Self { store, @@ -6151,6 +6224,7 @@ impl GcpConnectorResolver { .time_to_live(GCP_SETUP_REPORT_WINDOW) .max_capacity(10_000) .build(), + push_auth, } } @@ -6195,6 +6269,7 @@ impl GcpConnectorResolver { org_id: row.org_id, secret_key_id: cache_key.1.chars().take(16).collect(), logs_enabled: row.logs_enabled, + confirms_log_forwarding: false, }; self.cache.insert(cache_key, identity.clone()).await; @@ -6205,6 +6280,102 @@ impl GcpConnectorResolver { Ok(Some(identity.into_resolved(&routing))) } + /// A push Pub/Sub signed as the connector's push service account, for an + /// audience naming this connector. `Err` only when Google's keys can't be + /// fetched, which a redelivery may get past. + async fn resolve_connector_by_token( + &self, + connector_id: &str, + token: &str, + ) -> Result, String> { + let now_secs = u64::try_from(current_time_millis() / 1000).unwrap_or_default(); + let push = match self.push_auth.verify(token, now_secs).await { + Ok(push) => push, + Err(PushAuthError::Invalid(reason)) => { + debug!(connector_id, reason, "GCP push token refused"); + return Ok(None); + } + Err(PushAuthError::KeysUnavailable(error)) => return Err(error), + }; + // The audience is the receiver URL; its host is whatever the + // subscription was given, the path names the connector. + let audience_path = url::Url::parse(&push.audience) + .map(|url| url.path().to_owned()) + .unwrap_or_default(); + if audience_path != format!("/v1/logpush/gcp/{connector_id}") { + debug!(connector_id, "GCP push token is for another audience"); + return Ok(None); + } + + let cache_key = (connector_id.to_owned(), format!("oidc:{}", push.email)); + if let Some(identity) = self.cache.get(&cache_key).await { + let routing = self.routing.resolve_org_routing(&identity.org_id).await?; + return Ok(Some(identity.into_resolved(&routing))); + } + if self.negative_cache.contains_key(&cache_key) { + return Ok(None); + } + let row = self + .store + .fetch_gcp_connector_by_id(connector_id) + .await? + .filter(|row| push.email == gcp_push_service_account(connector_id, &row.project_id)); + let Some(row) = row else { + self.negative_cache.insert(cache_key, ()).await; + return Ok(None); + }; + + let routing = OrgRouting { + self_managed: row.self_managed, + clickhouse_ready: row.clickhouse_ready, + }; + let identity = GcpConnectorIdentity { + org_id: row.org_id, + // Names the signing account, as a secret's hash prefix names the secret. + secret_key_id: push.email.chars().take(16).collect(), + logs_enabled: row.logs_enabled, + confirms_log_forwarding: row.applied_logs_enabled.is_none(), + }; + self.cache.insert(cache_key, identity.clone()).await; + self.routing + .remember_org_routing(&identity.org_id, routing.clone()) + .await; + Ok(Some(identity.into_resolved(&routing))) + } + + /// Records log forwarding as set up, once. The cached identities stop + /// asking first, so a busy sink writes once rather than once per entry + /// until the write lands; a failed write lets a later push try again. + async fn confirm_log_forwarding(&self, connector_id: &str) { + // Collected first: the cache's iterator is not held across an await. + let pending: Vec<_> = self + .cache + .iter() + .filter(|(key, identity)| key.0 == connector_id && identity.confirms_log_forwarding) + .collect(); + for (key, identity) in &pending { + let confirmed = GcpConnectorIdentity { + confirms_log_forwarding: false, + ..identity.clone() + }; + self.cache.insert((**key).clone(), confirmed).await; + } + let report = gcp_logging::SetupReport { + logs: Some(true), + metrics: None, + }; + if let Err(error) = self + .store + .record_gcp_setup_report(connector_id, report, current_time_millis()) + .await + { + warn!(error = %error, connector_id, "Failed to record GCP log forwarding as set up"); + for (key, _) in &pending { + self.cache.invalidate(&**key).await; + } + } + } + /// Writes `last_received_at` (`error == None`) or `last_error`, at most /// once per connector and outcome per interval. Best-effort: the request's /// outcome is already decided, so a failed write is only logged. @@ -6374,6 +6545,27 @@ fn schema_revision_at_least(stamped: &str, needed: i32) -> bool { const SCHEMA_REVISION_COMPATIBLE_SQL: &str = "(CASE WHEN s.schema_version ~ '^[0-9]+$' THEN s.schema_version::int ELSE -1 END >= $1)"; +/// A `GcpConnectorRow`'s columns over `gcp_connectors c` joined to +/// `org_clickhouse_settings s`, with the schema revision as `$1`. +fn gcp_connector_columns() -> String { + format!( + "c.org_id, c.project_id, c.logs_enabled, c.applied_logs_enabled, \ + COALESCE(s.sync_status = 'connected', false) AS self_managed, \ + COALESCE(s.sync_status = 'connected' AND {SCHEMA_REVISION_COMPATIBLE_SQL}, false) AS clickhouse_ready" + ) +} + +fn gcp_connector_row(row: &tokio_postgres::Row) -> GcpConnectorRow { + GcpConnectorRow { + org_id: row.get("org_id"), + project_id: row.get("project_id"), + logs_enabled: row.get("logs_enabled"), + applied_logs_enabled: row.get("applied_logs_enabled"), + self_managed: row.get("self_managed"), + clickhouse_ready: row.get("clickhouse_ready"), + } +} + #[async_trait::async_trait] impl ClickHouseTargetProvider for ClickHouseTargetResolver { async fn resolve_clickhouse_target( @@ -6678,10 +6870,9 @@ impl KeyStore for PostgresKeyStore { let revision = required_schema_revision(); let client = self.client().await?; // No `enabled` column: deleting the row is how a connector is turned off. + let columns = gcp_connector_columns(); let sql = format!( - "SELECT c.org_id, c.logs_enabled, \ - COALESCE(s.sync_status = 'connected', false) AS self_managed, \ - COALESCE(s.sync_status = 'connected' AND {SCHEMA_REVISION_COMPATIBLE_SQL}, false) AS clickhouse_ready \ + "SELECT {columns} \ FROM gcp_connectors c \ LEFT JOIN org_clickhouse_settings s ON s.org_id = c.org_id \ WHERE c.id = $2 AND c.secret_hash = $3 LIMIT 1" @@ -6696,15 +6887,33 @@ impl KeyStore for PostgresKeyStore { )) .await .map_err(|error| format!("postgres fetch_gcp_connector failed: {error}"))?; - let Some(row) = rows.into_iter().next() else { - return Ok(None); - }; - Ok(Some(GcpConnectorRow { - org_id: row.get("org_id"), - logs_enabled: row.get("logs_enabled"), - self_managed: row.get("self_managed"), - clickhouse_ready: row.get("clickhouse_ready"), - })) + Ok(rows.first().map(gcp_connector_row)) + } + + async fn fetch_gcp_connector_by_id( + &self, + connector_id: &str, + ) -> Result, String> { + let revision = required_schema_revision(); + let client = self.client().await?; + let columns = gcp_connector_columns(); + let sql = format!( + "SELECT {columns} \ + FROM gcp_connectors c \ + LEFT JOIN org_clickhouse_settings s ON s.org_id = c.org_id \ + WHERE c.id = $2 LIMIT 1" + ); + let rows = client + .query(&sql, &[&revision, &connector_id]) + .instrument(postgres_client_span( + "fetch_gcp_connector_by_id", + "SELECT", + GCP_CONNECTORS_TABLE, + &self.target, + )) + .await + .map_err(|error| format!("postgres fetch_gcp_connector_by_id failed: {error}"))?; + Ok(rows.first().map(gcp_connector_row)) } async fn fetch_sampling_policy( @@ -6958,6 +7167,13 @@ impl KeyStore for StaticKeyStore { Ok(None) } + async fn fetch_gcp_connector_by_id( + &self, + _connector_id: &str, + ) -> Result, String> { + Ok(None) + } + async fn fetch_sampling_policy( &self, _org_id: &str, @@ -8212,6 +8428,18 @@ mod tests { .get(&(connector_id.to_owned(), secret_hash.to_owned())) .cloned()) } + async fn fetch_gcp_connector_by_id( + &self, + connector_id: &str, + ) -> Result, String> { + Ok(self + .gcp_connectors + .lock() + .unwrap() + .iter() + .find(|((id, _), _)| id == connector_id) + .map(|(_, row)| row.clone())) + } async fn fetch_sampling_policy( &self, _org_id: &str, @@ -8573,6 +8801,7 @@ mod tests { "test-hmac-key".to_owned(), routing, Duration::from_mins(1), + GooglePushAuth::with_keys(gcp_push_test_keys()), ), autumn_tracker: None, autumn_entitlements: None, @@ -9610,24 +9839,29 @@ mod tests { (url, rx) } - /// One org with two connectors: `gcp_conn_1`, and `gcp_conn_off` whose log - /// forwarding is turned off. + /// One org with three connectors: `gcp_conn_1`, `gcp_conn_off` whose log + /// forwarding is turned off, both set up by an earlier script that put the + /// secret in the push URL, and `GCP_SIGNED_CONNECTOR`, set up with + /// Terraform and never reported on, whose pushes Pub/Sub signs. async fn gcp_logpush_state( name: &str, clickhouse_url: &str, ) -> (Arc, Arc, PathBuf) { let queue_dir = unique_main_test_dir(name); let store = Arc::new(FakeKeyStore::default()); - for (connector_id, secret, logs_enabled) in [ - ("gcp_conn_1", "gcp-secret", true), - ("gcp_conn_off", "gcp-secret-off", false), + for (connector_id, secret, logs_enabled, applied_logs_enabled) in [ + ("gcp_conn_1", "gcp-secret", true, Some(true)), + ("gcp_conn_off", "gcp-secret-off", false, Some(true)), + (GCP_SIGNED_CONNECTOR, "gcp-secret-signed", true, None), ] { store.insert_gcp_connector( connector_id, secret, GcpConnectorRow { org_id: "org_gcp".to_owned(), + project_id: "acme-host".to_owned(), logs_enabled, + applied_logs_enabled, self_managed: true, clickhouse_ready: true, }, @@ -9667,11 +9901,154 @@ mod tests { Query(LogpushQuery { secret: secret.map(str::to_owned), }), + HeaderMap::new(), Bytes::from_static(body.as_bytes()), ) .await } + const GCP_SIGNED_CONNECTOR: &str = "0a1b2c3d-4e5f-4071-8293-a4b5c6d7e8f9"; + const GCP_SIGNED_ACCOUNT: &str = + "maple-push-0a1b2c3d4e5f4071829@acme-host.iam.gserviceaccount.com"; + const GCP_PUSH_TEST_KEY: &[u8] = include_bytes!("testdata/gcp-push-test-key.pk8"); + const GCP_PUSH_TEST_KEY_N: &str = include_str!("testdata/gcp-push-test-key.n"); + + fn gcp_push_test_keys() -> maple_ingest::gcp_push_auth::Keys { + maple_ingest::gcp_push_auth::Keys::from([( + "key-1".to_owned(), + maple_ingest::gcp_push_auth::RsaKey { + n: URL_SAFE_NO_PAD.decode(GCP_PUSH_TEST_KEY_N.trim()).unwrap(), + e: vec![1, 0, 1], + }, + )]) + } + + /// An ID token as Pub/Sub signs it for a push subscription's account. + fn gcp_push_token(email: &str, audience: &str) -> String { + use ring::signature::{RsaKeyPair, RSA_PKCS1_SHA256}; + let now = current_time_millis() / 1000; + let encode = |value: serde_json::Value| URL_SAFE_NO_PAD.encode(value.to_string()); + let signed = format!( + "{}.{}", + encode(serde_json::json!({ "alg": "RS256", "kid": "key-1", "typ": "JWT" })), + encode(serde_json::json!({ + "iss": "https://accounts.google.com", + "aud": audience, + "iat": now, + "exp": now + 3600, + "email": email, + "email_verified": true, + })), + ); + let pair = RsaKeyPair::from_pkcs8(GCP_PUSH_TEST_KEY).unwrap(); + let mut signature = vec![0; pair.public().modulus_len()]; + pair.sign( + &RSA_PKCS1_SHA256, + &ring::rand::SystemRandom::new(), + signed.as_bytes(), + &mut signature, + ) + .unwrap(); + format!("{signed}.{}", URL_SAFE_NO_PAD.encode(signature)) + } + + async fn gcp_signed_push( + state: &Arc, + connector_id: &str, + token: &str, + body: &'static str, + ) -> Response { + let mut headers = HeaderMap::new(); + headers.insert( + AUTHORIZATION, + HeaderValue::from_str(&format!("Bearer {token}")).unwrap(), + ); + handle_gcp_logpush( + State(Arc::clone(state)), + Path(connector_id.to_owned()), + Query(LogpushQuery { secret: None }), + headers, + Bytes::from_static(body.as_bytes()), + ) + .await + } + + #[tokio::test] + async fn gcp_logpush_accepts_a_push_signed_as_the_connectors_push_account() { + let (ch_url, mut ch_rx) = spawn_fake_clickhouse().await; + let (state, store, queue_dir) = gcp_logpush_state("gcp-logpush-signed", &ch_url).await; + let token = gcp_push_token( + GCP_SIGNED_ACCOUNT, + &format!("https://ingest.test/v1/logpush/gcp/{GCP_SIGNED_CONNECTOR}"), + ); + + for _ in 0..2 { + let response = + gcp_signed_push(&state, GCP_SIGNED_CONNECTOR, &token, GCP_LOG_ENTRY).await; + assert_eq!(response.status(), StatusCode::OK); + tokio::time::timeout(Duration::from_secs(2), ch_rx.recv()) + .await + .expect("the entry is stored") + .expect("fake ClickHouse is running"); + } + // Nothing reported on it before: the first push proves log forwarding + // is set up, and only the first records it. + assert_eq!( + *store.gcp_setup_reports.lock().unwrap(), + [( + GCP_SIGNED_CONNECTOR.to_owned(), + gcp_logging::SetupReport { + logs: Some(true), + metrics: None + } + )] + ); + + drop(std::fs::remove_dir_all(queue_dir)); + } + + #[tokio::test] + async fn gcp_logpush_refuses_a_token_for_another_account_or_connector() { + let (state, store, queue_dir) = + gcp_logpush_state("gcp-logpush-signed-refused", "http://127.0.0.1:1").await; + let audience = format!("https://ingest.test/v1/logpush/gcp/{GCP_SIGNED_CONNECTOR}"); + + for (case, connector_id, token) in [ + ( + "the same name in another project", + GCP_SIGNED_CONNECTOR, + gcp_push_token( + "maple-push-0a1b2c3d4e5f4071829@attacker.iam.gserviceaccount.com", + &audience, + ), + ), + ( + "a token for another connector's endpoint", + GCP_SIGNED_CONNECTOR, + gcp_push_token( + GCP_SIGNED_ACCOUNT, + "https://ingest.test/v1/logpush/gcp/gcp_conn_1", + ), + ), + ( + "another connector's endpoint with this token", + "gcp_conn_1", + gcp_push_token(GCP_SIGNED_ACCOUNT, &audience), + ), + ( + "not a token", + GCP_SIGNED_CONNECTOR, + "gcp-secret-signed".to_owned(), + ), + ] { + let response = gcp_signed_push(&state, connector_id, &token, GCP_LOG_ENTRY).await; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED, "{case}"); + } + assert!(store.gcp_setup_reports.lock().unwrap().is_empty()); + + drop(std::fs::remove_dir_all(queue_dir)); + } + #[tokio::test] async fn gcp_logpush_rejects_bad_credentials() { let (state, _store, queue_dir) = diff --git a/apps/ingest/src/testdata/gcp-push-test-key.n b/apps/ingest/src/testdata/gcp-push-test-key.n new file mode 100644 index 0000000000..423ab89096 --- /dev/null +++ b/apps/ingest/src/testdata/gcp-push-test-key.n @@ -0,0 +1 @@ +xw7kNmTxqS6r6IoeOEQW1iQ1oWnhL99UYQgthpdP8KgzW-FVhltnHG-X8tu8t-GIaC9j_Uhf9YH3268I0hpFbakiGNwy-WnctTy4UG5kY-S1RlTaTEgbIuI-N6BUoCCCbLQCBg3IP9e7O-c2fAZQ1pn7EAIV3qDtTGE4b4aBRrD3f6pJ-h8UgnhWn4u-1QzzhdPgRe22E5lceZyuMiFtZhzyxvak82FirJ5xGgJx-zVYpGDjEioczBLM-RR6LTFYo9Zk7wZugaGl0dgQfqwpQp3fMf0POjU0QDa5Dct76HvFf6CWAEUdwZpYlfdRJOOgZFe49bhmKhB_yTBOQpVocw \ No newline at end of file diff --git a/apps/ingest/src/testdata/gcp-push-test-key.pk8 b/apps/ingest/src/testdata/gcp-push-test-key.pk8 new file mode 100644 index 0000000000..d39622efe5 Binary files /dev/null and b/apps/ingest/src/testdata/gcp-push-test-key.pk8 differ diff --git a/apps/landing/src/content/docs/integrations/gcp.md b/apps/landing/src/content/docs/integrations/gcp.md index 54b6c97c9e..e5192d379b 100644 --- a/apps/landing/src/content/docs/integrations/gcp.md +++ b/apps/landing/src/content/docs/integrations/gcp.md @@ -70,8 +70,10 @@ Checking access ✓ Log filter accepted Log forwarding - ✓ APIs enabled (Pub/Sub, Cloud Logging, Cloud Resource Manager) + ✓ APIs enabled (Pub/Sub, Cloud Logging, IAM, Cloud Resource Manager) ✓ Topic created + ✓ Push service account created + ✓ Pub/Sub allowed to sign pushes as that account ✓ Push subscription created ✓ Log sink created ✓ Sink allowed to write to the project @@ -100,7 +102,7 @@ If a step fails, the script stops and prints what to do. Fix it and paste the sc | "Done. Google Cloud is set up for Maple." | Created the log sink or the read-only service account. | | "Done. Everything is in place." | Created and removed nothing. It makes every grant again, so it also restores one that was removed. | | "Done. The log sink has the filter of this script. Everything else is in place." | Replaced the sink's filter and created nothing. | -| "Done. Google Cloud matches your Maple switches." | Removed the log sink, topic and subscription: **Log forwarding** is off. | +| "Done. Google Cloud matches your Maple switches." | Removed the log sink, topic, subscription and push service account: **Log forwarding** is off. | After Maple confirms the run, **Log forwarding** and **Key metrics** each wait for their first data: @@ -109,9 +111,9 @@ After Maple confirms the run, **Log forwarding** and **Key metrics** each wait f | Log forwarding | A new sink can take about 10 minutes to start forwarding. Entries logged before that are not forwarded. | | Key metrics | The first read lands within about 10 minutes. Maple reads every 5 minutes, about 5 minutes behind, so that each minute is complete. | -The script contains a secret that lets anyone send logs to your Maple organization. Don't share it or commit it, and run it in Cloud Shell: there neither the paste nor the secret ends up in shell history. The pasted text keeps itself out of the history files of bash 5 and zsh, but it still shows in your terminal's scrollback. +Google Cloud never gets a secret: Pub/Sub signs each push as the connection's push service account, and Maple accepts a push only when Google signed it for that account and for this connection. The script itself contains a secret, which it uses only to report its run to Maple. Don't share it or commit it, and run it in Cloud Shell: there neither the paste nor the secret ends up in shell history. The pasted text keeps itself out of the history files of bash 5 and zsh, but it still shows in your terminal's scrollback. If it leaks, disconnect and connect again: the new connection gets a new secret. -The push endpoint with the secret is also visible in the subscription's configuration and in the host project's Admin Activity audit log, to anyone who can read those. If it leaks, disconnect and connect again: the new connection gets a new secret. +A connection set up before Maple signed its pushes has the secret in its push endpoint, visible in the subscription's configuration and the host project's Admin Activity audit log. Run the setup script again to switch it to signed pushes. The first two lines and the last two lines of the pasted text run the script in a bash process of its own, so a failed step can't close your Cloud Shell session. @@ -164,6 +166,44 @@ On a connection that already has a sink, the choice starts on **Keep current fil For any other filter, paste the script into an editor first, edit `LOG_FILTER` near its top and set `LOG_FILTER_MODE` to `set`. The filter uses the [Logging query language](https://cloud.google.com/logging/docs/view/logging-query-language). The script checks the filter with Google before it creates anything. Later runs with **Keep current filter** leave your filter in place, while **Replace filter** replaces it. +## Set up with Terraform + +Instead of the script, apply the same setup from your own Terraform (1.5 or later) or OpenTofu configuration. The module creates the resources the script does, under the same names, and holds no secret, so you can commit it. Its source is [`deploy/terraform-google-maple`](https://github.com/MapleTechLabs/maple/tree/main/deploy/terraform-google-maple) in Maple's repository. + +1. Create the connection in Maple as in steps 1 to 5 above. +2. In **Apply in Google Cloud**, choose **Terraform** and copy the module block into your configuration. +3. Run `terraform apply` as an account with the [roles for running the script](#roles-for-running-the-script). +4. Back in Maple, step 3 confirms each part: **Log forwarding** with the first log entry Maple receives, usually within 10 minutes, and **Read-only access** when you click **Check access**. + +A Terraform run sends Maple no report, which is why Maple checks Google Cloud itself. The module sets the sink's log filter in full: choose the filter in Maple before you copy the block. After changing **Log forwarding** in Maple, copy the block again and apply. + +To take over a connection the script set up, import its resources before the first apply. For a project connection, with the module named `maple_gcp_acme_prod` and the names from the script's header: + +```hcl +import { + to = module.maple_gcp_acme_prod.google_pubsub_topic.logs[0] + id = "projects/acme-prod/topics/maple-0a1b2c3d4e5f40718293a4b5" +} +import { + to = module.maple_gcp_acme_prod.google_pubsub_subscription.logs[0] + id = "projects/acme-prod/subscriptions/maple-0a1b2c3d4e5f40718293a4b5" +} +import { + to = module.maple_gcp_acme_prod.google_logging_project_sink.logs[0] + id = "projects/acme-prod/sinks/maple-0a1b2c3d4e5f40718293a4b5" +} +import { + to = module.maple_gcp_acme_prod.google_service_account.reader[0] + id = "projects/acme-prod/serviceAccounts/maple-0a1b2c3d4e5f40718293a4b5@acme-prod.iam.gserviceaccount.com" +} +import { + to = module.maple_gcp_acme_prod.google_service_account.push[0] + id = "projects/acme-prod/serviceAccounts/maple-push-0a1b2c3d4e5f4071829@acme-prod.iam.gserviceaccount.com" +} +``` + +A folder or organization connection imports `google_logging_folder_sink` from `folders/FOLDER_ID/sinks/NAME` or `google_logging_organization_sink` from `organizations/ORGANIZATION_ID/sinks/NAME`. Leave out the push service account if the script ran before Maple signed pushes. Role grants need no import: applying one that exists changes nothing. + ## Verify Once a connection has delivered data, **Integrations → Google Cloud** opens with a row of numbers, summed over every connection: @@ -224,16 +264,17 @@ Then check the data: The script enables these APIs in the host project: -- `pubsub.googleapis.com` and `logging.googleapis.com`, for log forwarding: carrying the log entries and routing them through the sink. +- `pubsub.googleapis.com`, `logging.googleapis.com` and `iam.googleapis.com`, for log forwarding: carrying the log entries, routing them through the sink, and the push service account. - `monitoring.googleapis.com`, `cloudasset.googleapis.com`, `iam.googleapis.com` and `iamcredentials.googleapis.com`, for the read-only service account: reading metrics, listing resources, creating the account and minting its short-lived tokens. No key is ever created. - `cloudresourcemanager.googleapis.com`, for both: granting the roles below. -For log forwarding it creates a Pub/Sub topic and push subscription in the host project and a log sink on the project, folder or organization. For every connection it creates a read-only service account in the host project, which Maple reads metrics as: live queries, and **Key metrics** while it is on. Each is named `maple-` followed by 24 hexadecimal characters unique to the connection. Running the script again never removes the service account, whatever the switches say; only the [cleanup script](#disconnect) does. +For log forwarding it creates a Pub/Sub topic, a push subscription and a push service account in the host project, and a log sink on the project, folder or organization. The push service account, `maple-push-` followed by 19 hexadecimal characters, holds no role anywhere: Pub/Sub signs each push to Maple as it. For every connection it creates a read-only service account in the host project, which Maple reads metrics as: live queries, and **Key metrics** while it is on. Each is named `maple-` followed by 24 hexadecimal characters unique to the connection. Running the script again never removes the service account, whatever the switches say; only the [cleanup script](#disconnect) does. It grants these roles: - **Pub/Sub Publisher** (`roles/pubsub.publisher`) to the sink's Google-managed writer identity, on Maple's topic: publishing log entries to the topic. - **Logs Writer** (`roles/logging.logWriter`) to the same writer identity, on the host project: Google requires it on the project that holds a sink's destination. +- **Service Account Token Creator** (`roles/iam.serviceAccountTokenCreator`) to the host project's Pub/Sub service agent, on the push service account: signing pushes as it. Projects created after April 2021 allow this by default. - **Monitoring Viewer** (`roles/monitoring.viewer`) to the read-only service account, on the project, folder or organization: reading metrics (`monitoring.timeSeries.list`), live and every 5 minutes. - **Cloud Asset Viewer** (`roles/cloudasset.viewer`) to the read-only service account, on the project, folder or organization: listing resources and the projects under a folder or organization (`cloudasset.assets.searchAllResources`). The role can also read resource metadata and IAM policies. - **Service Usage Consumer** (`roles/serviceusage.serviceUsageConsumer`) to the read-only service account, on the host project: calling both read APIs through the host project. @@ -245,14 +286,15 @@ It grants these roles: Owner on the host project is the simplest. Without it, these predefined roles together cover what the script does: -| On | Role | For | -| ----------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ----------------------------------------------- | -| The host project | Service Usage Admin (`roles/serviceusage.serviceUsageAdmin`) | Switching on the APIs above. | -| The host project | Project IAM Admin (`roles/resourcemanager.projectIamAdmin`) | The role grants on the host project. | -| The host project | Pub/Sub Admin (`roles/pubsub.admin`) | Log forwarding: the topic and the subscription. | -| The host project | Service Account Admin (`roles/iam.serviceAccountAdmin`) | The read-only service account. | -| The project, folder or organization | Logs Configuration Writer (`roles/logging.configWriter`) | Log forwarding: the sink. | -| The project, folder or organization | Project IAM Admin (`roles/resourcemanager.projectIamAdmin`), Folder IAM Admin (`roles/resourcemanager.folderIamAdmin`) or Organization Administrator (`roles/resourcemanager.organizationAdmin`) | The read-only roles. | +| On | Role | For | +| ----------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------- | +| The host project | Service Usage Admin (`roles/serviceusage.serviceUsageAdmin`) | Switching on the APIs above. | +| The host project | Project IAM Admin (`roles/resourcemanager.projectIamAdmin`) | The role grants on the host project. | +| The host project | Pub/Sub Admin (`roles/pubsub.admin`) | Log forwarding: the topic and the subscription. | +| The host project | Service Account Admin (`roles/iam.serviceAccountAdmin`) | The read-only and push service accounts. | +| The host project | Service Account User (`roles/iam.serviceAccountUser`) | Log forwarding: signing pushes as the push service account. | +| The project, folder or organization | Logs Configuration Writer (`roles/logging.configWriter`) | Log forwarding: the sink. | +| The project, folder or organization | Project IAM Admin (`roles/resourcemanager.projectIamAdmin`), Folder IAM Admin (`roles/resourcemanager.folderIamAdmin`) or Organization Administrator (`roles/resourcemanager.organizationAdmin`) | The read-only roles. | Before it changes anything, the script asks Google which permissions the signed-in account holds and stops with the missing ones. Read-only access asks for its own just before its section: what it lacks stops only that section while log forwarding is on, and the run ends with "Done, except read-only access". @@ -595,12 +637,12 @@ Google checks the policy when a grant is made, so leave the exception in place: ## Disconnect 1. On **Integrations → Google Cloud**, open the **⋮** menu (**More actions**) on the connection's row and choose **Disconnect**. -2. Click **Copy cleanup script** and run it in Cloud Shell. It deletes the log sink, topic, subscription and read-only service account, and ends with "Done. Everything the setup script created is gone." +2. Click **Copy cleanup script** and run it in Cloud Shell. It deletes the log sink, topic, subscription and both service accounts, and ends with "Done. Everything the setup script created is gone." Set up with Terraform: remove the module and apply instead. 3. Wait for the check mark in the dialog, then click **Disconnect**. Maple stops reading the connection's metrics, and stops accepting its logs within about a minute. Data already in Maple is kept. The cleanup takes one to two minutes when the connection forwards logs: after it deletes the sink it waits a minute for Google to stop routing to the topic, so your project logs no sink error. Turning **Log forwarding** off in **Configure** and running the setup script takes as long, for the same reason. -**Disconnect anyway** disconnects without the cleanup. Google Cloud then keeps publishing logs to Pub/Sub, billed by Google, until the cleanup script runs. Maple keeps a panel with **Copy cleanup script** on the page until you click **Dismiss** and confirm. +**Disconnect anyway** disconnects without the cleanup, and is the button to use after removing the module with Terraform, which reports nothing to Maple. Google Cloud then keeps publishing logs to Pub/Sub, billed by Google, until the cleanup script runs. Maple keeps a panel with **Copy cleanup script** on the page until you click **Dismiss** and confirm. The cleanup script leaves the APIs it switched on enabled, and the Logs Writer role of Google's logging service account on the host project, which other sinks share. diff --git a/apps/web/src/components/integrations/gcp-configure.tsx b/apps/web/src/components/integrations/gcp-configure.tsx index 44e4b9fa32..0982028bc9 100644 --- a/apps/web/src/components/integrations/gcp-configure.tsx +++ b/apps/web/src/components/integrations/gcp-configure.tsx @@ -24,6 +24,7 @@ import { Input } from "@maple/ui/components/ui/input" import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@maple/ui/components/ui/select" import { Skeleton } from "@maple/ui/components/ui/skeleton" import { Spinner } from "@maple/ui/components/ui/spinner" +import { ToggleGroup, ToggleGroupItem } from "@maple/ui/components/ui/toggle-group" import { countLabel } from "@maple/ui/lib/format" import { cn } from "@maple/ui/lib/utils" @@ -344,6 +345,53 @@ function Script({ ) } +/** + * The Terraform module block: short enough to show whole, and free of secrets, so it reads as + * what it is before anyone copies it. + */ +function TerraformBlock({ + terraform, + failure, +}: { + terraform: string | null + failure: { readonly message: string; readonly retry: () => void; readonly retrying: boolean } | null +}) { + const text = terraform?.trimEnd() ?? null + if (failure !== null) { + return ( +
+

+ {failure.message} +

+ +
+ ) + } + return text === null ? ( + + ) : ( +
+
+ Terraform + +
+ 
+				{text}
+
+
+ ) +} + type StepState = "current" | "pending" | "waiting" | "done" /** @@ -1133,6 +1181,20 @@ function ApplyStage({ const scopeRoles = gcpScopeRoles(connector.scope_type, connector.logs_enabled, metricsAvailable) const scopeName = GCP_SCOPE_NAMES[connector.scope_type].toLowerCase() const role = (name: string) => {name} + const [method, setMethod] = useState<"script" | "terraform">("script") + const whoRuns = ( + <> + {role("Owner")} of {connector.scope_type === "project" ? null : "the host project "} + {connector.project_id} + {scopeRoles.map((name, index) => ( + + {index === 0 ? " who also has " : " and "} + {role(name)} + + ))} + {scopeRoles.length > 0 ? ` on the ${scopeName}` : null} + + ) return ( <> @@ -1175,88 +1237,114 @@ function ApplyStage({ ) : ( -
    - - Sign in as an {role("Owner")} of{" "} - {connector.scope_type === "project" ? null : "the host project "} - {connector.project_id} - {scopeRoles.map((name, index) => ( - - {index === 0 ? " who also has " : " and "} - {role(name)} - - ))} - {scopeRoles.length > 0 ? ` on the ${scopeName}` : null}.{" "} - - Roles if you aren't an Owner - - - } - > -
    - -
    -     - -