diff --git a/.env.example b/.env.example index 933ac5a..f167953 100644 --- a/.env.example +++ b/.env.example @@ -16,3 +16,7 @@ DNS_CACHE_SIZE=262144 DEDUP_SIZE=262144 LOG_LEVEL=info ACTIVE_PROBING=false + +# Optional HTTP basic auth for the web UI (/ and /api/*): +# UI_USER=admin +# UI_PASS=change-me-strong diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index ce63b5d..ea62dca 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -46,7 +46,7 @@ Certificate Transparency Logs | Direct queue | bound, non-blocking send | unbounded growth | | DNS enrichment | DNS via worker pool, TTL cache, CNAME follow | unbounded goroutines, persistence | | Dedup | memory-bounded TTL caches | unbounded maps | -| Probe (optional) | DNS/TCP/TLS/HTTP HEAD/GET `/` | auth, downloads, mutation, DB or shell access | +| Probe (optional) | one unauthenticated vendor read endpoint (S3/GCS/Azure/Firebase list) or a bare TCP connect for databases | credentials, object downloads, mutation, DB queries, auth attempts | | DB writer | batched `pgx.Batch` writes | one synchronous row per event | | Checkpoints | JSON CT position in `ct_checkpoints`/`ct_sths` | certificate/issuer contents in DB | diff --git a/README.md b/README.md index 4a2d0a3..2f3a11d 100644 --- a/README.md +++ b/README.md @@ -121,9 +121,11 @@ Environment variables only; no config files, no runtime pattern files. | `DB_BATCH_SIZE` | `500` | findings per write batch | | `DB_FLUSH_INTERVAL` | `2s` | max time before a batch is flushed | | `ACTIVE_PROBING` | `false` | enable optional reachability checks | -| `PROBE_WORKERS` / `PROBE_TIMEOUT` | `4` / `5s` | probe pool | +| `PROBE_WORKERS` / `PROBE_TIMEOUT` | `8` / `5s` | probe worker pool | +| `PROBE_QUEUE_SIZE` | `65536` | probe sample size; when full, old entries are replaced at random (bounded memory) | | `DEDUP_TTL` | `1h` | hostname/finding dedup window | -| `METRICS_ADDR` | `:9090` | metrics/health listener | +| `METRICS_ADDR` | `:9090` | metrics/health/UI listener | +| `UI_USER` / `UI_PASS` | *(unset)* | optional HTTP basic auth for the web UI | | `LOG_LEVEL` | `info` | `debug\|info\|warn\|error` | | `STATS_INTERVAL` | `60s` | aggregated stats log interval | | `LOG_LIST_SOURCE` | Google log list v3 | certspotter log list URL | @@ -142,9 +144,44 @@ container runs as non-root, has a read-only root filesystem, drops all Linux capabilities, sets CPU/memory/pids limits and mounts no application volume. The only persistent state is PostgreSQL (CT checkpoints + cloud assets). -Endpoints: `GET /metrics`, `GET /healthz`, `GET /readyz` on port `9090`, bound -to `127.0.0.1` only. The listener has no authentication, so scrape it locally or -from a sidecar and never expose it publicly. +Endpoints on port `9090`, bound to `127.0.0.1` only: + +* `GET /` — web UI (see below) +* `GET /api/assets`, `GET /api/stats` — JSON for the UI +* `GET /metrics`, `GET /healthz`, `GET /readyz` + +The listener has no authentication by default, so scrape it locally or from a +sidecar and never expose it publicly. Set `UI_USER`/`UI_PASS` to require HTTP +basic auth on the UI and its API. + +## Web UI + +A small, dependency-free UI lists discovered assets with provider/service, +resource, region, exposure state and the probe detail, plus summary counts and +provider/service/exposure filters. It is read-only, refreshes every 15s, and is +served from the same loopback port (`http://127.0.0.1:9090/`). + +## Reachability and readability checks + +`ACTIVE_PROBING=false` by default. When enabled, the prober issues **one** +unauthenticated request per finding. It never sends credentials, downloads +objects, mutates anything, runs database queries or attempts authentication. + +| Vendor | Check | `public_read` means | +| --- | --- | --- | +| AWS S3, Cloudflare R2, DigitalOcean Spaces | anonymous `ListObjectsV2` (`?list-type=2&max-keys=1`) | anonymous bucket listing succeeded | +| Google Cloud Storage | JSON API object list (`/storage/v1/b//o?maxResults=1`) | `allUsers` can list objects | +| Azure Blob/File/Queue/Table | anonymous `?comp=list` | anonymous listing succeeded | +| Firebase Realtime DB | `GET /.json?shallow=true` | security rules allow anonymous read | +| Supabase | `GET /rest/v1/` | — (an API key is required, so this reports `auth_required`) | +| Cosmos, OpenSearch, Search, Appspot, OCI | root `GET /` | a 2xx was returned | +| RDS, Redshift, DocumentDB, Neptune, ElastiCache, MemoryDB, Azure SQL/Postgres/MySQL/Redis, Supabase Postgres | unauthenticated TCP connect only | *never* — reported as `reachable` (port open) | + +Probe outcomes are stored in `exposure_state` and a short factual +`exposure_detail` string (e.g. `s3 redirect region=eu-central-1`, +`status=403 authentication required`, `tcp/5432 open; no authentication +attempted`). Database endpoints are never claimed to be readable: a raw +`public_read` on a database is impossible by design. ## Production notes @@ -167,8 +204,8 @@ from a sidecar and never expose it publicly. * `cloud_assets` — one row per `(provider, service, provider_hostname)`, with `hostname` (original CT hostname), `resource_identifier`, `region`, - `resource_specific`, `public_semantic`, `exposure_state`, `first_seen`, - `last_seen`. + `resource_specific`, `public_semantic`, `exposure_state`, `exposure_detail`, + `probed_at`, `first_seen`, `last_seen`. * `observations` — append-only observations linking an asset to a `source_hostname`, `source_type` (`direct_ct`/`cname`), CNAME chain and A/AAAA records. diff --git a/cmd/ctcloud/main.go b/cmd/ctcloud/main.go index 18dcd17..321a69e 100644 --- a/cmd/ctcloud/main.go +++ b/cmd/ctcloud/main.go @@ -24,6 +24,7 @@ import ( "tea.karlcom.de/karl/bucketeer/internal/pipeline" "tea.karlcom.de/karl/bucketeer/internal/probe" "tea.karlcom.de/karl/bucketeer/internal/store" + "tea.karlcom.de/karl/bucketeer/internal/webui" ) func main() { @@ -87,9 +88,10 @@ func run() error { prober = probe.New(probe.Config{ Workers: cfg.ProbeWorkers, Timeout: cfg.ProbeTimeout, + Queue: cfg.ProbeQueueSize, Metrics: m, - Update: func(uctx context.Context, c model.Candidate, state model.ExposureState) { - _ = st.UpdateExposure(uctx, c.Provider.String(), c.Service.String(), c.Hostname, state) + Update: func(uctx context.Context, c model.Candidate, state model.ExposureState, detail string) { + _ = st.UpdateExposure(uctx, c.Provider.String(), c.Service.String(), c.Hostname, state, detail) }, }) prober.Start(ctx) @@ -137,7 +139,8 @@ func run() error { monitorDone <- monitor.Run(ctx, monitorCfg) }() - httpSrv := startHTTP(cfg.MetricsAddr, m, st, &monitorReady) + ui := webui.New(st, webui.Config{User: cfg.UIUser, Pass: cfg.UIPass}) + httpSrv := startHTTP(cfg.MetricsAddr, m, st, ui, &monitorReady) go logStats(ctx, m, cfg.StatsInterval) slog.Info("ctcloud started") @@ -172,9 +175,10 @@ func run() error { return nil } -func startHTTP(addr string, m *metrics.Metrics, st *store.Store, ready *atomic.Bool) *http.Server { +func startHTTP(addr string, m *metrics.Metrics, st *store.Store, ui http.Handler, ready *atomic.Bool) *http.Server { mux := http.NewServeMux() mux.Handle("/metrics", m.Handler()) + mux.Handle("/", ui) mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("ok\n")) diff --git a/internal/config/config.go b/internal/config/config.go index 9cdbcc4..6fdbc0e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -31,11 +31,14 @@ type Config struct { DBBatchSize int DBFlushInterval time.Duration - ActiveProbing bool - ProbeWorkers int - ProbeTimeout time.Duration + ActiveProbing bool + ProbeWorkers int + ProbeTimeout time.Duration + ProbeQueueSize int MetricsAddr string + UIUser string + UIPass string LogLevel string StatsInterval time.Duration @@ -70,9 +73,12 @@ func Load() (*Config, error) { DBBatchSize: envInt("DB_BATCH_SIZE", 500), DBFlushInterval: envDuration("DB_FLUSH_INTERVAL", 2*time.Second), ActiveProbing: envBool("ACTIVE_PROBING", false), - ProbeWorkers: envInt("PROBE_WORKERS", 4), + ProbeWorkers: envInt("PROBE_WORKERS", 8), ProbeTimeout: envDuration("PROBE_TIMEOUT", 5*time.Second), + ProbeQueueSize: envInt("PROBE_QUEUE_SIZE", 1<<16), MetricsAddr: envString("METRICS_ADDR", ":9090"), + UIUser: envString("UI_USER", ""), + UIPass: envString("UI_PASS", ""), LogLevel: envString("LOG_LEVEL", "info"), StatsInterval: envDuration("STATS_INTERVAL", 60*time.Second), LogListSource: envString("LOG_LIST_SOURCE", DefaultLogListSource), diff --git a/internal/probe/probe.go b/internal/probe/probe.go index 82c6d9e..da40ea5 100644 --- a/internal/probe/probe.go +++ b/internal/probe/probe.go @@ -1,6 +1,13 @@ -// Package probe performs optional, minimal, non-destructive reachability checks. -// It never authenticates, downloads objects, mutates anything, or runs database -// queries; it only observes the response to a single HTTPS GET /. +// Package probe performs optional, minimal, non-destructive reachability and +// readability checks. It only ever issues an unauthenticated HTTP request or an +// unauthenticated TCP connect. It never sends credentials, downloads objects, +// mutates anything, runs database queries, or attempts authentication. +// +// "readable" means an anonymous request was answered with data: for object +// stores we use the vendor's anonymous list endpoint, for Firebase the REST +// data endpoint, etc. For database endpoints there is no anonymous read concept, +// so the best honest signal is whether the endpoint is publicly resolvable and +// its port is open; we report that as reachable, never as readable. package probe import ( @@ -8,8 +15,11 @@ import ( "crypto/tls" "errors" "fmt" + "io" + mathrand "math/rand/v2" "net" "net/http" + "net/url" "sync" "time" @@ -17,8 +27,8 @@ import ( "tea.karlcom.de/karl/bucketeer/internal/model" ) -// UpdateFunc records the observed exposure state. -type UpdateFunc func(ctx context.Context, c model.Candidate, state model.ExposureState) +// UpdateFunc records the observed exposure state and a short factual detail. +type UpdateFunc func(ctx context.Context, c model.Candidate, state model.ExposureState, detail string) type Config struct { Workers int @@ -28,14 +38,58 @@ type Config struct { Update UpdateFunc } -// Prober runs a fixed pool of workers over a bounded queue. type Prober struct { cfg Config - queue chan model.Candidate + sample *sampleQueue client *http.Client wg sync.WaitGroup } +// sampleQueue is a memory-bounded queue that keeps a random sample of recent +// submissions rather than a FIFO. With FIFO, one high-volume hostname template +// (e.g. *.mrap.accesspoint.s3-global.amazonaws.com) fills the entire queue and +// no other asset type is ever probed. Random replacement means every template +// gets sampled over time. It is not on the certificate hot path. +type sampleQueue struct { + mu sync.Mutex + items []model.Candidate + cap int + notify chan struct{} +} + +func newSampleQueue(capacity int) *sampleQueue { + return &sampleQueue{cap: capacity, notify: make(chan struct{}, 1)} +} + +func (q *sampleQueue) push(c model.Candidate) { + q.mu.Lock() + if len(q.items) < q.cap { + q.items = append(q.items, c) + } else { + q.items[mathrand.IntN(len(q.items))] = c + } + q.mu.Unlock() + select { + case q.notify <- struct{}{}: + default: + } +} + +func (q *sampleQueue) pop() (model.Candidate, bool) { + q.mu.Lock() + n := len(q.items) + if n == 0 { + q.mu.Unlock() + return model.Candidate{}, false + } + i := mathrand.IntN(n) + c := q.items[i] + q.items[i] = q.items[n-1] + q.items = q.items[:n-1] + q.mu.Unlock() + return c, true +} + func New(cfg Config) *Prober { if cfg.Workers <= 0 { cfg.Workers = 4 @@ -47,10 +101,12 @@ func New(cfg Config) *Prober { cfg.Queue = 1024 } return &Prober{ - cfg: cfg, - queue: make(chan model.Candidate, cfg.Queue), + cfg: cfg, + sample: newSampleQueue(cfg.Queue), client: &http.Client{ Timeout: cfg.Timeout, + // Observe the first response; do not follow redirects so region + // hints (e.g. S3 301 PermanentRedirect) stay visible. CheckRedirect: func(req *http.Request, via []*http.Request) error { return http.ErrUseLastResponse }, @@ -75,15 +131,14 @@ func (p *Prober) Start(ctx context.Context) { } } -// Submit enqueues a candidate without blocking; it drops when the queue is full. +// Submit enqueues a candidate without blocking. When the sample is full the +// oldest-slot candidate is replaced at random; this is load shedding, not an +// error, so it is not counted as a drop. func (p *Prober) Submit(c model.Candidate) { if p.cfg.Metrics != nil { p.cfg.Metrics.ProbeAttempts.Add(1) } - select { - case p.queue <- c: - default: - } + p.sample.push(c) } func (p *Prober) Wait() { p.wg.Wait() } @@ -94,46 +149,196 @@ func (p *Prober) worker(ctx context.Context) { select { case <-ctx.Done(): return - case c, ok := <-p.queue: + case <-p.sample.notify: + } + for { + c, ok := p.sample.pop() if !ok { - return + break } - state := p.probe(ctx, c) + state, detail := p.Probe(ctx, c) if p.cfg.Metrics != nil && state == model.ExposurePublicRead { p.cfg.Metrics.ProbePublic.Add(1) } if p.cfg.Update != nil { - p.cfg.Update(ctx, c, state) + p.cfg.Update(ctx, c, state, detail) } } } } -func (p *Prober) probe(ctx context.Context, c model.Candidate) model.ExposureState { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+c.Hostname+"/", nil) +// Probe runs the vendor-appropriate check for a single candidate. +func (p *Prober) Probe(ctx context.Context, c model.Candidate) (model.ExposureState, string) { + switch c.Provider { + case model.ProviderAWS: + switch c.Service { + case model.ServiceS3: + return p.checkS3(ctx, c) + case model.ServiceRDS, model.ServiceDocumentDB: + return p.checkTCP(ctx, c.Hostname, 5432) + case model.ServiceNeptune: + return p.checkTCP(ctx, c.Hostname, 8182) + case model.ServiceElastiCache, model.ServiceMemoryDB: + return p.checkTCP(ctx, c.Hostname, 6379) + case model.ServiceRedshift: + return p.checkTCP(ctx, c.Hostname, 5439) + case model.ServiceOpenSearch: + return p.checkHTTP(ctx, "https://"+c.Hostname+"/") + } + case model.ProviderAzure: + switch c.Service { + case model.ServiceBlobStorage, model.ServiceFileStorage, model.ServiceQueue, model.ServiceTableStorage: + // Anonymous container/file/share/queue/table listing. + return p.checkHTTP(ctx, "https://"+c.Hostname+"/?comp=list") + case model.ServiceWebApp, model.ServiceSearch: + return p.checkHTTP(ctx, "https://"+c.Hostname+"/") + case model.ServiceCosmosDB: + return p.checkHTTP(ctx, "https://"+c.Hostname+"/") + case model.ServiceDatabase: // Azure SQL + return p.checkTCP(ctx, c.Hostname, 1433) + case model.ServicePostgres: + return p.checkTCP(ctx, c.Hostname, 5432) + case model.ServiceMySQL: + return p.checkTCP(ctx, c.Hostname, 3306) + case model.ServiceRedis: + return p.checkTCP(ctx, c.Hostname, 6380) + } + case model.ProviderGCP: + switch c.Service { + case model.ServiceObjectStorage: + return p.checkGCS(ctx, c) + case model.ServiceFirebaseDB: + return p.checkFirebase(ctx, c) + case model.ServiceWebApp: + return p.checkHTTP(ctx, "https://"+c.Hostname+"/") + } + case model.ProviderCloudflare: + return p.checkS3(ctx, c) // R2 is S3-compatible (and r2.dev serves objects) + case model.ProviderDigitalOcean: + if c.Service == model.ServiceObjectStorage { + return p.checkS3(ctx, c) + } + return p.checkTCP(ctx, c.Hostname, 5432) + case model.ProviderSupabase: + if c.Service == model.ServicePostgres { + return p.checkTCP(ctx, c.Hostname, 5432) + } + return p.checkHTTP(ctx, "https://"+c.Hostname+"/rest/v1/") + case model.ProviderOracle: + return p.checkHTTP(ctx, "https://"+c.Hostname+"/") + } + return p.checkHTTP(ctx, "https://"+c.Hostname+"/") +} + +// checkS3 queries the S3 ListObjectsV2 anonymous endpoint. A 200 means the +// bucket allows anonymous listing (public read); 403 means it exists but is not +// public; 404 means no such bucket. 301/307 carries the real region. +func (p *Prober) checkS3(ctx context.Context, c model.Candidate) (model.ExposureState, string) { + status, header, err := p.do(ctx, "https://"+c.Hostname+"/?list-type=2&max-keys=1") if err != nil { - return model.ExposureUnknown + return classifyError(err) + } + if status == http.StatusMovedPermanently || status == http.StatusTemporaryRedirect || status == http.StatusPermanentRedirect { + if region := header.Get("x-amz-bucket-region"); region != "" { + return model.ExposureReachable, "s3 redirect region=" + region + } + return model.ExposureReachable, fmt.Sprintf("s3 redirect status=%d", status) + } + return classifyStatus(status, "s3") +} + +// checkGCS uses the public JSON API to test object listing, which requires the +// bucket to grant allUsers read access. +func (p *Prober) checkGCS(ctx context.Context, c model.Candidate) (model.ExposureState, string) { + bucket := c.Resource + if bucket == "" { + bucket = c.Hostname + } + api := "https://storage.googleapis.com/storage/v1/b/" + url.PathEscape(bucket) + "/o?maxResults=1" + status, _, err := p.do(ctx, api) + if err != nil { + return classifyError(err) + } + return classifyStatus(status, "gcs") +} + +// checkFirebase queries the Realtime Database REST endpoint with shallow=true so +// only keys are returned. 200 means the security rules allow anonymous reads. +func (p *Prober) checkFirebase(ctx context.Context, c model.Candidate) (model.ExposureState, string) { + status, _, err := p.do(ctx, "https://"+c.Hostname+"/.json?shallow=true") + if err != nil { + return classifyError(err) + } + return classifyStatus(status, "firebase") +} + +func (p *Prober) checkHTTP(ctx context.Context, endpoint string) (model.ExposureState, string) { + status, _, err := p.do(ctx, endpoint) + if err != nil { + return classifyError(err) + } + return classifyStatus(status, "") +} + +// checkTCP reports whether a database endpoint is publicly resolvable and its +// port is open. It intentionally does not attempt any protocol handshake or +// authentication, so it can never be reported as readable. +func (p *Prober) checkTCP(ctx context.Context, host string, port int) (model.ExposureState, string) { + d := net.Dialer{Timeout: p.cfg.Timeout} + conn, err := d.DialContext(ctx, "tcp", net.JoinHostPort(host, fmt.Sprint(port))) + if err != nil { + return classifyError(err) + } + _ = conn.Close() + return model.ExposureReachable, fmt.Sprintf("tcp/%d open; no authentication attempted", port) +} + +// do issues a single GET and reads at most a tiny prefix of the body before +// closing it, so no object is ever downloaded. +func (p *Prober) do(ctx context.Context, endpoint string) (int, http.Header, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return 0, nil, err } resp, err := p.client.Do(req) if err != nil { - var dnsErr *net.DNSError - if errors.As(err, &dnsErr) { - return model.ExposurePrivate - } - return model.ExposureUnknown + return 0, nil, err } defer resp.Body.Close() + _, _ = io.CopyN(io.Discard, resp.Body, 512) + return resp.StatusCode, resp.Header, nil +} - switch { - case resp.StatusCode >= 200 && resp.StatusCode < 300: - return model.ExposurePublicRead - case resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusUnauthorized: - return model.ExposureAuthRequired - case resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone: - return model.ExposureNotFound - default: - return model.ExposureReachable +func classifyStatus(status int, tag string) (model.ExposureState, string) { + prefix := "" + if tag != "" { + prefix = tag + " " } + switch { + case status >= 200 && status < 300: + return model.ExposurePublicRead, fmt.Sprintf("%sstatus=%d anonymous listing allowed", prefix, status) + case status == http.StatusUnauthorized || status == http.StatusForbidden: + return model.ExposureAuthRequired, fmt.Sprintf("%sstatus=%d authentication required", prefix, status) + case status == http.StatusNotFound || status == http.StatusGone: + return model.ExposureNotFound, fmt.Sprintf("%sstatus=%d not found", prefix, status) + default: + return model.ExposureReachable, fmt.Sprintf("%sstatus=%d", prefix, status) + } +} + +func classifyError(err error) (model.ExposureState, string) { + var dnsErr *net.DNSError + if errors.As(err, &dnsErr) { + if dnsErr.IsNotFound { + return model.ExposurePrivate, "dns: no public record" + } + return model.ExposureUnknown, "dns: " + dnsErr.Err + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return model.ExposureUnknown, "timeout" + } + return model.ExposureUnknown, err.Error() } // Describe is a helper for logs/tests. diff --git a/internal/probe/probe_test.go b/internal/probe/probe_test.go index 69a0200..a70ac29 100644 --- a/internal/probe/probe_test.go +++ b/internal/probe/probe_test.go @@ -1,7 +1,9 @@ package probe import ( - "crypto/tls" + "context" + "errors" + "net" "net/http" "net/http/httptest" "strings" @@ -11,20 +13,18 @@ import ( "tea.karlcom.de/karl/bucketeer/internal/model" ) -func newProber(t *testing.T, status int) (*Prober, string) { +func serverHost(t *testing.T, h http.Handler) string { t.Helper() - srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(status) - })) + srv := httptest.NewTLSServer(h) t.Cleanup(srv.Close) - p := New(Config{Workers: 1, Timeout: time.Second}) - // httptest uses a self-signed cert; the prober already skips verification. - _ = tls.Config{} - host := strings.TrimPrefix(srv.URL, "https://") - return p, host + return strings.TrimPrefix(srv.URL, "https://") } -func TestProbeClassification(t *testing.T) { +func statusHandler(status int) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(status) }) +} + +func TestProbeStorageStatuses(t *testing.T) { cases := []struct { status int want model.ExposureState @@ -35,11 +35,97 @@ func TestProbeClassification(t *testing.T) { {http.StatusNotFound, model.ExposureNotFound}, {http.StatusInternalServerError, model.ExposureReachable}, } - for _, c := range cases { - p, host := newProber(t, c.status) - got := p.probe(t.Context(), model.Candidate{Hostname: host}) - if got != c.want { - t.Errorf("status %d: got %s, want %s", c.status, got, c.want) + for _, tc := range cases { + p := New(Config{Workers: 1, Timeout: time.Second}) + host := serverHost(t, statusHandler(tc.status)) + got, detail := p.Probe(context.Background(), model.Candidate{ + Hostname: host, Provider: model.ProviderAWS, Service: model.ServiceS3, + }) + if got != tc.want { + t.Errorf("status %d: got %s (%s), want %s", tc.status, got, detail, tc.want) } } } + +func TestProbeS3RegionRedirect(t *testing.T) { + h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("x-amz-bucket-region", "eu-central-1") + w.WriteHeader(http.StatusMovedPermanently) + }) + p := New(Config{Workers: 1, Timeout: time.Second}) + host := serverHost(t, h) + got, detail := p.Probe(context.Background(), model.Candidate{ + Hostname: host, Provider: model.ProviderAWS, Service: model.ServiceS3, + }) + if got != model.ExposureReachable || !strings.Contains(detail, "eu-central-1") { + t.Fatalf("got %s (%s), want reachable with region", got, detail) + } +} + +func TestProbeFirebase(t *testing.T) { + p := New(Config{Workers: 1, Timeout: time.Second}) + host := serverHost(t, statusHandler(http.StatusUnauthorized)) + got, _ := p.Probe(context.Background(), model.Candidate{ + Hostname: host, Provider: model.ProviderGCP, Service: model.ServiceFirebaseDB, + }) + if got != model.ExposureAuthRequired { + t.Fatalf("got %s, want auth_required", got) + } + host2 := serverHost(t, statusHandler(http.StatusOK)) + got, _ = p.Probe(context.Background(), model.Candidate{ + Hostname: host2, Provider: model.ProviderGCP, Service: model.ServiceFirebaseDB, + }) + if got != model.ExposurePublicRead { + t.Fatalf("got %s, want public_read", got) + } +} + +func TestProbeDatabaseTCP(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + conn.Close() + } + }() + host, port, _ := net.SplitHostPort(ln.Addr().String()) + p := New(Config{Workers: 1, Timeout: time.Second}) + state, detail := p.checkTCP(context.Background(), host, mustAtoi(port)) + if state != model.ExposureReachable { + t.Fatalf("got %s (%s), want reachable", state, detail) + } + if !strings.Contains(detail, "no authentication attempted") { + t.Fatalf("detail should be explicit about no auth: %q", detail) + } + + // A closed port yields unknown, never readable. + state, _ = p.checkTCP(context.Background(), "127.0.0.1", 1) + if state == model.ExposurePublicRead { + t.Fatal("closed port must never be public_read") + } +} + +func TestClassifyErrorNXDOMAIN(t *testing.T) { + state, detail := classifyError(&net.DNSError{Err: "no such host", IsNotFound: true}) + if state != model.ExposurePrivate || !strings.Contains(detail, "no public record") { + t.Fatalf("got %s (%s)", state, detail) + } + if state, _ := classifyError(errors.New("boom")); state != model.ExposureUnknown { + t.Fatalf("got %s, want unknown", state) + } +} + +func mustAtoi(s string) int { + n := 0 + for _, c := range s { + n = n*10 + int(c-'0') + } + return n +} diff --git a/internal/store/store.go b/internal/store/store.go index 809e3e9..81a0934 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -138,9 +138,120 @@ func nonNil(s []string) []string { } // UpdateExposure records the outcome of an optional reachability probe. -func (s *Store) UpdateExposure(ctx context.Context, provider, service, providerHostname string, state model.ExposureState) error { - _, err := s.pool.Exec(ctx, - `UPDATE cloud_assets SET exposure_state = $4 WHERE provider = $1 AND service = $2 AND provider_hostname = $3`, - provider, service, providerHostname, state.String()) +func (s *Store) UpdateExposure(ctx context.Context, provider, service, providerHostname string, state model.ExposureState, detail string) error { + _, err := s.pool.Exec(ctx, ` + UPDATE cloud_assets + SET exposure_state = $4, exposure_detail = $5, probed_at = now() + WHERE provider = $1 AND service = $2 AND provider_hostname = $3`, + provider, service, providerHostname, state.String(), detail) return err } + +// Asset is a row of the cloud_assets table for the read-only UI. +type Asset struct { + ID int64 `json:"id"` + Provider string `json:"provider"` + Service string `json:"service"` + Hostname string `json:"hostname"` + ProviderHost string `json:"provider_hostname"` + Resource string `json:"resource_identifier"` + Region string `json:"region"` + PublicSemantic string `json:"public_semantic"` + Exposure string `json:"exposure_state"` + ExposureDetail string `json:"exposure_detail"` + FirstSeen time.Time `json:"first_seen"` + LastSeen time.Time `json:"last_seen"` + ProbedAt *time.Time `json:"probed_at"` +} + +// AssetFilter selects assets for the UI. Empty fields are ignored. +type AssetFilter struct { + Provider string + Service string + Exposure string + Query string + Limit int + Offset int +} + +// ListAssets returns assets matching the filter, newest first. +func (s *Store) ListAssets(ctx context.Context, f AssetFilter) ([]Asset, error) { + if f.Limit <= 0 || f.Limit > 500 { + f.Limit = 100 + } + rows, err := s.pool.Query(ctx, ` + SELECT id, provider, service, hostname, provider_hostname, resource_identifier, + region, public_semantic, exposure_state, exposure_detail, first_seen, last_seen, probed_at + FROM cloud_assets + WHERE ($1 = '' OR provider = $1) + AND ($2 = '' OR service = $2) + AND ($3 = '' OR exposure_state = $3) + AND ($4 = '' OR provider_hostname ILIKE '%' || $4 || '%' OR hostname ILIKE '%' || $4 || '%' OR resource_identifier ILIKE '%' || $4 || '%') + ORDER BY last_seen DESC + LIMIT $5 OFFSET $6`, + f.Provider, f.Service, f.Exposure, f.Query, f.Limit, f.Offset) + if err != nil { + return nil, err + } + defer rows.Close() + out := []Asset{} + for rows.Next() { + var a Asset + if err := rows.Scan(&a.ID, &a.Provider, &a.Service, &a.Hostname, &a.ProviderHost, + &a.Resource, &a.Region, &a.PublicSemantic, &a.Exposure, &a.ExposureDetail, + &a.FirstSeen, &a.LastSeen, &a.ProbedAt); err != nil { + return nil, err + } + out = append(out, a) + } + return out, rows.Err() +} + +// Count is a (label, count) pair for the UI summary. +type Count struct { + Label string `json:"label"` + Count int64 `json:"count"` +} + +// Stats summarizes the assets table. +type Stats struct { + Total int64 `json:"total"` + Readable int64 `json:"readable"` + ByProvider []Count `json:"by_provider"` + ByService []Count `json:"by_service"` + ByExposure []Count `json:"by_exposure"` +} + +func (s *Store) Stats(ctx context.Context) (Stats, error) { + var st Stats + if err := s.pool.QueryRow(ctx, `SELECT count(*), count(*) FILTER (WHERE exposure_state = 'public_read') FROM cloud_assets`).Scan(&st.Total, &st.Readable); err != nil { + return st, err + } + group := func(column string) ([]Count, error) { + rows, err := s.pool.Query(ctx, `SELECT `+column+`, count(*) FROM cloud_assets GROUP BY 1 ORDER BY 2 DESC`) + if err != nil { + return nil, err + } + defer rows.Close() + var counts []Count + for rows.Next() { + var c Count + if err := rows.Scan(&c.Label, &c.Count); err != nil { + return nil, err + } + counts = append(counts, c) + } + return counts, rows.Err() + } + var err error + if st.ByProvider, err = group("provider"); err != nil { + return st, err + } + if st.ByService, err = group("service"); err != nil { + return st, err + } + if st.ByExposure, err = group("exposure_state"); err != nil { + return st, err + } + return st, nil +} diff --git a/internal/store/store_test.go b/internal/store/store_test.go index d8eeeda..f5a6742 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -60,7 +60,13 @@ func TestIntegration(t *testing.T) { if err := s.WriteFindings(ctx, []pipeline.Finding{f}); err != nil { t.Fatal(err) } - if err := s.UpdateExposure(ctx, model.ProviderAWS.String(), model.ServiceS3.String(), f.Hostname, model.ExposureReachable); err != nil { + if err := s.UpdateExposure(ctx, model.ProviderAWS.String(), model.ServiceS3.String(), f.Hostname, model.ExposureReachable, "test"); err != nil { + t.Fatal(err) + } + if _, err := s.ListAssets(ctx, AssetFilter{Provider: "aws", Limit: 10}); err != nil { + t.Fatal(err) + } + if _, err := s.Stats(ctx); err != nil { t.Fatal(err) } } diff --git a/internal/webui/index.html b/internal/webui/index.html new file mode 100644 index 0000000..8d02b21 --- /dev/null +++ b/internal/webui/index.html @@ -0,0 +1,123 @@ + + + + + +ctcloud · cloud assets + + + +
+

ctcloud

+ cloud/database resources discovered in Certificate Transparency + — +
+
+
+
+ + + + + +
+ + + + + + +
providerserviceresourceregionhostnamehostname (CT)exposuredetaillast seen
+
+ + + + diff --git a/internal/webui/webui.go b/internal/webui/webui.go new file mode 100644 index 0000000..23a4258 --- /dev/null +++ b/internal/webui/webui.go @@ -0,0 +1,90 @@ +// Package webui serves a small read-only web UI over the discovered cloud +// assets. It is intentionally dependency-free (stdlib + embedded HTML) and +// exposes no mutation endpoints. +package webui + +import ( + "context" + _ "embed" + "encoding/json" + "net/http" + "strconv" + "strings" + + "tea.karlcom.de/karl/bucketeer/internal/store" +) + +//go:embed index.html +var indexHTML []byte + +// Store is the read-only subset of the database the UI needs. +type Store interface { + ListAssets(ctx context.Context, f store.AssetFilter) ([]store.Asset, error) + Stats(ctx context.Context) (store.Stats, error) +} + +type Config struct { + User string // optional HTTP basic auth username + Pass string // optional HTTP basic auth password +} + +// New returns the UI handler. If Config.User is set, all routes require basic auth. +func New(s Store, cfg Config) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + _, _ = w.Write(indexHTML) + }) + mux.HandleFunc("/api/assets", func(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + limit, _ := strconv.Atoi(q.Get("limit")) + offset, _ := strconv.Atoi(q.Get("offset")) + assets, err := s.ListAssets(r.Context(), store.AssetFilter{ + Provider: strings.TrimSpace(q.Get("provider")), + Service: strings.TrimSpace(q.Get("service")), + Exposure: strings.TrimSpace(q.Get("exposure")), + Query: strings.TrimSpace(q.Get("q")), + Limit: limit, + Offset: offset, + }) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + writeJSON(w, assets) + }) + mux.HandleFunc("/api/stats", func(w http.ResponseWriter, r *http.Request) { + st, err := s.Stats(r.Context()) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + writeJSON(w, st) + }) + + if cfg.User != "" { + return basicAuth(cfg.User, cfg.Pass, mux) + } + return mux +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + _ = json.NewEncoder(w).Encode(v) +} + +func basicAuth(user, pass string, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + u, p, ok := r.BasicAuth() + if !ok || u != user || p != pass { + w.Header().Set("WWW-Authenticate", `Basic realm="ctcloud"`) + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + next.ServeHTTP(w, r) + }) +} diff --git a/internal/webui/webui_test.go b/internal/webui/webui_test.go new file mode 100644 index 0000000..c17b544 --- /dev/null +++ b/internal/webui/webui_test.go @@ -0,0 +1,94 @@ +package webui + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "tea.karlcom.de/karl/bucketeer/internal/store" +) + +type fakeStore struct { + lastFilter store.AssetFilter +} + +func (f *fakeStore) ListAssets(ctx context.Context, fl store.AssetFilter) ([]store.Asset, error) { + f.lastFilter = fl + return []store.Asset{{ID: 1, Provider: "aws", Service: "s3", ProviderHost: "b.s3.amazonaws.com", Exposure: "public_read"}}, nil +} + +func (f *fakeStore) Stats(ctx context.Context) (store.Stats, error) { + return store.Stats{Total: 1, Readable: 1}, nil +} + +func TestAPI(t *testing.T) { + fs := &fakeStore{} + srv := httptest.NewServer(New(fs, Config{})) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/api/assets?provider=aws&q=bucket&limit=5") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d", resp.StatusCode) + } + var assets []store.Asset + if err := json.NewDecoder(resp.Body).Decode(&assets); err != nil { + t.Fatal(err) + } + if len(assets) != 1 || fs.lastFilter.Provider != "aws" || fs.lastFilter.Query != "bucket" || fs.lastFilter.Limit != 5 { + t.Fatalf("unexpected assets/filter: %+v %+v", assets, fs.lastFilter) + } + + statsResp, err := http.Get(srv.URL + "/api/stats") + if err != nil { + t.Fatal(err) + } + defer statsResp.Body.Close() + var st store.Stats + if err := json.NewDecoder(statsResp.Body).Decode(&st); err != nil { + t.Fatal(err) + } + if st.Total != 1 { + t.Fatalf("stats = %+v", st) + } +} + +func TestBasicAuth(t *testing.T) { + srv := httptest.NewServer(New(&fakeStore{}, Config{User: "u", Pass: "p"})) + defer srv.Close() + + resp, _ := http.Get(srv.URL + "/api/stats") + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401", resp.StatusCode) + } + resp.Body.Close() + + req, _ := http.NewRequest(http.MethodGet, srv.URL+"/api/stats", nil) + req.SetBasicAuth("u", "p") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } +} + +func TestIndexServed(t *testing.T) { + srv := httptest.NewServer(New(&fakeStore{}, Config{})) + defer srv.Close() + resp, err := http.Get(srv.URL + "/") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if ct := resp.Header.Get("Content-Type"); ct != "text/html; charset=utf-8" { + t.Fatalf("content-type = %q", ct) + } +} diff --git a/migrations/0001_schema.sql b/migrations/0001_schema.sql index 2d67eda..2ed8eb3 100644 --- a/migrations/0001_schema.sql +++ b/migrations/0001_schema.sql @@ -19,6 +19,11 @@ CREATE TABLE IF NOT EXISTS cloud_assets ( CREATE INDEX IF NOT EXISTS cloud_assets_last_seen_idx ON cloud_assets (last_seen); +-- Reachability probe results (added after the initial schema; ALTER keeps this +-- idempotent for existing databases). +ALTER TABLE cloud_assets ADD COLUMN IF NOT EXISTS exposure_detail TEXT NOT NULL DEFAULT ''; +ALTER TABLE cloud_assets ADD COLUMN IF NOT EXISTS probed_at TIMESTAMPTZ; + CREATE TABLE IF NOT EXISTS observations ( id BIGSERIAL PRIMARY KEY, asset_id BIGINT NOT NULL REFERENCES cloud_assets (id) ON DELETE CASCADE,