jcoffey-dev
17fdc212c2
Ingest tenant-awareness was named "undesigned, not just unbuilt" across CLAUDE.md/threat-model.md/the runbook since early Phase 4 -- the last major standing gap. Scoping was agreed via AskUserQuestion: a config-supplied tenant_id + shared-secret token ingest validates (smaller real implementation, no new PKI), over per-tenant mTLS certs. This change builds that identity mechanism end to end and attaches it to every record at the point it enters the system; it deliberately does NOT build per-tenant write-routing for ClickHouse or Tantivy -- that's real, separately-scoped follow-up work, disclosed explicitly everywhere this was previously called undesigned, not silently left half-done. New pieces: - metadata/migrations/0034 + enterprise/internal/rbacstore/ ingest_credentials.go: a per-tenant bearer credential, only its SHA-256 hash ever persisted (same reasoning a password gets hashed, not stored raw) -- CreateIngestCredential returns the plaintext exactly once, ValidateIngestCredential/RevokeIngestCredential/ ListIngestCredentialsForTenant round it out. - enterprise-auth gains -create-ingest-credential-tenant/ -list-ingest-credentials-tenant/-revoke-ingest-credential (same offline-operator-flag shape as every other credential-minting flag in this binary) and a new POST /internal/authorize-ingest endpoint (internal/authhandler) validating a presented token and resolving its tenant -- a genuinely different credential type from session-backed /internal/authorize, so it doesn't touch session.Manager at all. - ingest (AGPL core) gains an optional TenantResolver (internal/grpcserver, nil by default) and its HTTP client implementation (internal/tenantresolver.HTTPResolver) -- a plain HTTP call to enterprise-auth's new endpoint, never an enterprise/ import, same "network boundary, not import boundary" shape api/authz.HTTPAuthorizer already uses for the query path. PushBatch now requires an `authorization: Bearer <token>` gRPC metadata entry once a resolver is configured, fails the whole batch closed on a missing/invalid credential (never falls back to "no tenant"), and attaches the resolved tenant ID to every record as a `tenant_id` Kafka message header before producing it. Verified with real round trips at every layer, no Docker needed: rbacstore's credential CRUD (skip-gated on live Postgres, same as every other rbacstore integration test this phase), authhandler's new endpoint (real HTTP via httptest, including the regression test that a session token must not validate as an ingest credential), tenantresolver (real HTTP client against httptest, same pattern as authz.HTTPAuthorizer's own tests), and grpcserver's PushBatch (fake resolver/producer -- no resolver leaves messages unchanged, a configured resolver attaches the right header or fails closed on a bad/missing token). Helm: ingest.requireTenantCredential (default false) is a deliberate, separate opt-in from enterprise.enabled -- turning ENTERPRISE_AUTH_URL on for ingest requires every agent to already hold a credential or be refused outright, so it must not default on just because enterprise.enabled does (same reasoning api.yaml's ENTERPRISE_AUTH_URL isn't tied to enterprise.enabled directly either). docker-compose.yml leaves it unset, same as ever. Docs updated everywhere this was called "undesigned": CLAUDE.md, docs/architecture.md, docs/security/threat-model.md (including its summary table, now split into "identity: built" vs "write-routing: not yet"), docs/phase-4-runbook.md (new §13), enterprise/README.md.
103 lines
2.7 KiB
Go
103 lines
2.7 KiB
Go
// Package config loads ingest's configuration from environment variables.
|
|
// Phase 0 deliberately has no config file format of its own — env vars are
|
|
// enough for a docker-compose/k8s deployment and avoid pulling in a config
|
|
// library.
|
|
package config
|
|
|
|
import (
|
|
"fmt"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
)
|
|
|
|
type Config struct {
|
|
GRPC GRPCConfig
|
|
TLS TLSConfig
|
|
Redpanda RedpandaConfig
|
|
ClickHouse ClickHouseConfig
|
|
Batch BatchConfig
|
|
// EnterpriseAuthURL enables per-tenant ingest credential validation
|
|
// (internal/grpcserver.TenantResolver) when set -- empty (the
|
|
// default) is a documented no-op, same "off unless configured" shape
|
|
// as every other optional enterprise integration point in this
|
|
// codebase (e.g. api's own ENTERPRISE_AUTH_URL).
|
|
EnterpriseAuthURL string
|
|
}
|
|
|
|
type GRPCConfig struct {
|
|
ListenAddr string
|
|
}
|
|
|
|
// TLSConfig is the server-side mTLS material: the ingest service's own
|
|
// cert/key, and the CA used to verify agent client certs.
|
|
type TLSConfig struct {
|
|
CertFile string
|
|
KeyFile string
|
|
ClientCAFile string
|
|
}
|
|
|
|
type RedpandaConfig struct {
|
|
Brokers []string
|
|
Topic string
|
|
ConsumerGroup string
|
|
}
|
|
|
|
type ClickHouseConfig struct {
|
|
Addr string
|
|
Database string
|
|
Username string
|
|
Password string
|
|
}
|
|
|
|
type BatchConfig struct {
|
|
MaxSize int
|
|
FlushIntervalMS int
|
|
}
|
|
|
|
func Load() (Config, error) {
|
|
cfg := Config{
|
|
GRPC: GRPCConfig{
|
|
ListenAddr: getenv("GRPC_LISTEN_ADDR", ":4317"),
|
|
},
|
|
TLS: TLSConfig{
|
|
CertFile: getenv("TLS_CERT_FILE", "/etc/sentry-ingest/server.pem"),
|
|
KeyFile: getenv("TLS_KEY_FILE", "/etc/sentry-ingest/server-key.pem"),
|
|
ClientCAFile: getenv("TLS_CLIENT_CA_FILE", "/etc/sentry-ingest/ca.pem"),
|
|
},
|
|
Redpanda: RedpandaConfig{
|
|
Brokers: strings.Split(getenv("REDPANDA_BROKERS", "localhost:9092"), ","),
|
|
Topic: getenv("REDPANDA_TOPIC", "sentry.logs.raw"),
|
|
ConsumerGroup: getenv("REDPANDA_CONSUMER_GROUP", "sentry-ingest"),
|
|
},
|
|
ClickHouse: ClickHouseConfig{
|
|
Addr: getenv("CLICKHOUSE_ADDR", "localhost:9000"),
|
|
Database: getenv("CLICKHOUSE_DATABASE", "sentry"),
|
|
Username: getenv("CLICKHOUSE_USERNAME", "default"),
|
|
Password: getenv("CLICKHOUSE_PASSWORD", ""),
|
|
},
|
|
EnterpriseAuthURL: getenv("ENTERPRISE_AUTH_URL", ""),
|
|
}
|
|
|
|
maxSize, err := strconv.Atoi(getenv("CONSUMER_BATCH_MAX_SIZE", "500"))
|
|
if err != nil {
|
|
return Config{}, fmt.Errorf("CONSUMER_BATCH_MAX_SIZE: %w", err)
|
|
}
|
|
cfg.Batch.MaxSize = maxSize
|
|
|
|
flushMS, err := strconv.Atoi(getenv("CONSUMER_BATCH_FLUSH_INTERVAL_MS", "2000"))
|
|
if err != nil {
|
|
return Config{}, fmt.Errorf("CONSUMER_BATCH_FLUSH_INTERVAL_MS: %w", err)
|
|
}
|
|
cfg.Batch.FlushIntervalMS = flushMS
|
|
|
|
return cfg, nil
|
|
}
|
|
|
|
func getenv(key, fallback string) string {
|
|
if v := os.Getenv(key); v != "" {
|
|
return v
|
|
}
|
|
return fallback
|
|
}
|