From 73109641e8bdd82adaeffbf8f8bb8532ee09741e Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Wed, 4 Feb 2026 10:25:09 -0600 Subject: [PATCH] add scan reports to hold pds --- CLAUDE.md | 208 +++++++---- pkg/atproto/cbor_gen.go | 583 ++++++++++++++++++++++++++++++- pkg/atproto/generate.go | 1 + pkg/atproto/lexicon.go | 57 ++- pkg/hold/pds/scan.go | 59 ++++ pkg/hold/pds/scan_broadcaster.go | 146 ++------ pkg/hold/pds/server.go | 1 + 7 files changed, 864 insertions(+), 191 deletions(-) create mode 100644 pkg/hold/pds/scan.go diff --git a/CLAUDE.md b/CLAUDE.md index 7caf237..c1b80ba 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -6,16 +6,29 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co ATCR (ATProto Container Registry) is an OCI-compliant container registry that uses the AT Protocol for manifest storage and S3 for blob storage. This creates a decentralized container registry where manifests are stored in users' Personal Data Servers (PDS) while layers are stored in S3. +## Go Workspace + +The project uses a Go workspace (`go.work`) with two modules: +- `atcr.io` — Main module (appview, hold, credential-helper, oauth-helper) +- `atcr.io/scanner` — Scanner module (separate to isolate heavy Syft/Grype dependencies) + ## Build Commands +Always build into the `bin/` directory (`-o bin/...`), not the project root. + ```bash -# Build all binaries -# create go builds in the bin/ directory +# Build main binaries go build -o bin/atcr-appview ./cmd/appview go build -o bin/atcr-hold ./cmd/hold go build -o bin/docker-credential-atcr ./cmd/credential-helper go build -o bin/oauth-helper ./cmd/oauth-helper +# Build scanner (separate module) +cd scanner && go build -o ../bin/atcr-scanner ./cmd/scanner && cd .. + +# Build hold with billing support (optional, uses build tag) +go build -tags billing -o bin/atcr-hold ./cmd/hold + # Run tests go test ./... @@ -38,40 +51,36 @@ go mod tidy # Build Docker images docker build -t atcr.io/appview:latest . docker build -f Dockerfile.hold -t atcr.io/hold:latest . +docker build -f Dockerfile.scanner -t atcr.io/scanner:latest . # Or use docker-compose docker-compose up -d -# Run locally (AppView) - configure via env vars (see .env.appview.example) -export ATCR_HTTP_ADDR=:5000 -export ATCR_DEFAULT_HOLD=http://127.0.0.1:8080 -./bin/atcr-appview serve +# Generate default config files +./bin/atcr-appview config init config-appview.yaml +./bin/atcr-hold config init config-hold.yaml -# Or use .env file: -cp .env.appview.example .env.appview -# Edit .env.appview with your settings -source .env.appview -./bin/atcr-appview serve +# Run locally (AppView) - YAML config (preferred) +./bin/atcr-appview serve --config config-appview.yaml +# Or env vars only (still works): +ATCR_SERVER_DEFAULT_HOLD_DID=did:web:hold01.atcr.io ./bin/atcr-appview serve -# Legacy mode (still supported): -# ./bin/atcr-appview serve config/config.yml - -# Run hold service (configure via env vars - see .env.hold.example) +# Run hold service - YAML config (preferred) # For local development, use Minio as S3-compatible storage: # docker run -p 9000:9000 minio/minio server /data -export HOLD_PUBLIC_URL=http://127.0.0.1:8080 -export AWS_ACCESS_KEY_ID=minioadmin -export AWS_SECRET_ACCESS_KEY=minioadmin -export S3_BUCKET=test -export S3_ENDPOINT=http://localhost:9000 -export HOLD_OWNER=did:plc:your-did-here -./bin/atcr-hold -# Hold starts immediately with embedded PDS +./bin/atcr-hold serve --config config-hold.yaml +# Or env vars only: +HOLD_SERVER_PUBLIC_URL=http://127.0.0.1:8080 S3_BUCKET=test ./bin/atcr-hold serve + +# Run scanner service (env vars only, no YAML) +SCANNER_HOLD_URL=ws://localhost:8080 SCANNER_SHARED_SECRET=secret ./bin/atcr-scanner serve + +# Usage report tool +go run ./cmd/usage-report --hold https://hold01.atcr.io +go run ./cmd/usage-report --hold https://hold01.atcr.io --from-manifests # Request Bluesky relay crawl (makes your PDS discoverable) ./deploy/request-crawl.sh hold01.atcr.io -# Or specify a different relay: -./deploy/request-crawl.sh hold01.atcr.io https://custom-relay.example.com/xrpc/com.atproto.sync.requestCrawl ``` ## Architecture Overview @@ -84,7 +93,7 @@ ATCR uses **distribution/distribution** as a library and extends it through midd - **Blobs/Layers** → S3 or user-deployed storage (large binary data) - **Authentication** → ATProto OAuth with DPoP + Docker credential helpers -### Three-Component Architecture +### Four-Component Architecture 1. **AppView** (`cmd/appview`) - OCI Distribution API server - Resolves identities (handle/DID → PDS endpoint) @@ -99,9 +108,17 @@ ATCR uses **distribution/distribution** as a library and extends it through midd - Supports S3-compatible storage (AWS S3, Storj, Minio, UpCloud, etc.) - Authorization based on captain record (public, allowAllCrew) - Self-describing via DID resolution - - Configured entirely via environment variables + - Optional subsystems: admin UI, quota enforcement, billing (Stripe), garbage collection + - Dispatches scan jobs to scanner instances via WebSocket -3. **Credential Helper** (`cmd/credential-helper`) - Client-side OAuth +3. **Scanner** (`scanner/cmd/scanner`) - Vulnerability scanning service + - Separate Go module (heavy Syft/Grype dependencies isolated) + - Connects to hold service via WebSocket (`/xrpc/io.atcr.hold.subscribeScanJobs`) + - Generates SBOMs with Syft, scans for vulnerabilities with Grype + - Priority queue with tier-based scheduling (owner > quartermaster > bosun > deckhand) + - Competing-consumer pattern: multiple scanners pull from same hold + +4. **Credential Helper** (`cmd/credential-helper`) - Client-side OAuth - Implements Docker credential helper protocol - ATProto OAuth flow with DPoP - Token caching and refresh @@ -559,19 +576,44 @@ All require blob:write permission via service token authentication: - Hold validates tokens and checks crew membership for authorization - Tokens cached for 50 seconds (valid for 60 seconds from PDS) -**Configuration:** Environment variables (see `.env.hold.example`) -- `HOLD_PUBLIC_URL` - Public URL of hold service (required, used for did:web generation) -- `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY` - S3 credentials (required) -- `S3_BUCKET` - S3 bucket name (required) -- `S3_ENDPOINT` - S3 endpoint URL (for non-AWS providers like Storj, Minio, UpCloud) -- `HOLD_PUBLIC` - Allow public reads (default: false) -- `HOLD_OWNER` - DID for captain record creation (optional) -- `HOLD_ALLOW_ALL_CREW` - Allow any authenticated user to register as crew (default: false) -- `HOLD_DATABASE_DIR` - Directory for embedded PDS database (required) -- `HOLD_KEY_PATH` - Path for PDS signing keys (optional, generated if missing) +**Hold Subsystems** (`pkg/hold/`): +- **Admin UI** (`admin/`) - Web-based admin panel for crew and storage management (enabled via `admin.enabled: true`) +- **Quota** (`quota/`) - Per-user storage quota enforcement with tier-based limits (configured in YAML under `quota:`) +- **Billing** (`billing/`) - Stripe integration for paid tiers (build tag `billing`, compile with `-tags billing`). Zero overhead when disabled. +- **Garbage Collection** (`gc/`) - Blob garbage collection for orphaned data +- **Scan Broadcaster** (`pds/scan_broadcaster.go`) - WebSocket server dispatching scan jobs to scanner instances via `/xrpc/io.atcr.hold.subscribeScanJobs` **Deployment:** Can run on Fly.io, Railway, Docker, Kubernetes, etc. +#### Scanner Service (`scanner/`) + +Separate Go module for vulnerability scanning. Connects to hold services via WebSocket. + +**Architecture:** +- `scanner/internal/client/hold.go` - WebSocket client with auto-reconnect (exponential backoff) +- `scanner/internal/queue/priority_queue.go` - Thread-safe priority queue (tier-based: owner > quartermaster > bosun > deckhand) +- `scanner/internal/scan/worker.go` - Configurable worker pool +- `scanner/internal/scan/syft.go` - SBOM generation via Syft +- `scanner/internal/scan/grype.go` - Vulnerability scanning via Grype +- `scanner/internal/scan/extractor.go` - Container layer extraction +- `scanner/internal/config/config.go` - Environment-only config (no YAML, no Viper) + +**Scanner env vars** (prefix `SCANNER_`): +- `SCANNER_HOLD_URL` - WebSocket URL of hold service (required) +- `SCANNER_SHARED_SECRET` - Authentication secret (required) +- `SCANNER_WORKERS` - Number of concurrent scan workers +- `SCANNER_VULN_ENABLED` - Enable vulnerability scanning (default: true) +- `SCANNER_ADDR` - Health endpoint address (default: `:9090`) + +#### Usage Report Tool (`cmd/usage-report/`) + +CLI tool for analyzing hold storage usage: +```bash +go run ./cmd/usage-report --hold https://hold01.atcr.io # summary +go run ./cmd/usage-report --hold https://hold01.atcr.io --from-manifests # from manifests +go run ./cmd/usage-report --hold https://hold01.atcr.io --list-blobs # individual blobs +``` + ### ATProto Storage Model Manifests are stored as records with this structure: @@ -657,39 +699,56 @@ This ensures: ### Configuration -**AppView configuration** (environment variables): +ATCR uses **Viper** for configuration. YAML files are the primary method; environment variables work as overrides. -Both AppView and Hold service follow the same pattern: **zero config files, all configuration via environment variables**. +**Loading priority** (highest wins): +1. Environment variables (always override YAML) +2. YAML config file (via `--config` / `-c` flag) +3. Hardcoded defaults -See `.env.appview.example` for all available options. Key environment variables: +**Generating config files:** +```bash +./bin/atcr-appview config init config-appview.yaml # fully-commented YAML with defaults +./bin/atcr-hold config init config-hold.yaml +``` -**Server:** -- `ATCR_HTTP_ADDR` - HTTP listen address (default: `:5000`) -- `ATCR_BASE_URL` - Public URL for OAuth/JWT realm (auto-detected in dev) -- `ATCR_DEFAULT_HOLD_DID` - Default hold DID for blob storage (REQUIRED, e.g., `did:web:hold01.atcr.io`) +**Env var naming convention:** Prefix + YAML path with `_` separators: +- AppView prefix: `ATCR_` — e.g., `server.default_hold_did` → `ATCR_SERVER_DEFAULT_HOLD_DID` +- Hold prefix: `HOLD_` — e.g., `server.public_url` → `HOLD_SERVER_PUBLIC_URL` +- S3 uses standard AWS names: `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `S3_BUCKET`, `S3_ENDPOINT` -**Authentication:** -- `ATCR_AUTH_KEY_PATH` - JWT signing key path (default: `/var/lib/atcr/auth/private-key.pem`) +**AppView config** (see `config-appview.example.yaml`): +- `server.addr` - Listen address (default: `:5000`) +- `server.base_url` - Public URL for OAuth/JWT realm (auto-detected in dev) +- `server.default_hold_did` - Default hold DID for blob storage (REQUIRED) +- `server.oauth_key_path` - P-256 key for OAuth client auth (auto-generated) +- `server.registry_domain` - Separate domain for OCI API (e.g., `buoy.cr`) +- `auth.key_path` - RSA key for signing registry JWTs +- `ui.database_path` - SQLite database path +- `jetstream.url` - ATProto firehose endpoint +- `jetstream.backfill_enabled` - Sync existing records on startup +- `log_shipper` - Remote log shipping (victoria, opensearch, loki) -**UI:** -- `ATCR_UI_DATABASE_PATH` - SQLite database path (default: `/var/lib/atcr/ui.db`) +**Hold config** (see `config-hold.example.yaml`): +- `server.public_url` - Externally reachable URL for did:web identity (REQUIRED) +- `server.public` - Allow unauthenticated reads (default: false) +- `storage.bucket` - S3 bucket (REQUIRED) +- `storage.endpoint` - Custom S3 endpoint for non-AWS providers +- `registration.owner_did` - DID for captain record auto-creation +- `registration.allow_all_crew` - Allow any authenticated user to join +- `database.path` - Embedded PDS database directory +- `admin.enabled` - Enable web admin panel +- `quota.tiers` - Storage quota tiers (e.g., `deckhand: {quota: "5GB"}`) +- `quota.defaults.new_crew_tier` - Default tier for new crew members -**Jetstream:** -- `JETSTREAM_URL` - ATProto event stream URL -- `ATCR_BACKFILL_ENABLED` - Enable periodic sync (default: false) +**Hold billing config** (requires `-tags billing` build): +- `billing.enabled` - Enable Stripe billing +- `billing.currency` - ISO currency code +- `billing.tiers` - Map of tier names to Stripe price IDs +- Env vars: `STRIPE_SECRET_KEY`, `STRIPE_WEBHOOK_SECRET` -**Legacy:** `config/config.yml` is still supported but deprecated. Use environment variables instead. - -**Hold Service configuration** (environment variables): - -See `.env.hold.example` for all available options. Key environment variables: -- `HOLD_PUBLIC_URL` - Public URL of hold service (REQUIRED) -- `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY` - S3 credentials (REQUIRED) -- `S3_BUCKET` - S3 bucket name (REQUIRED) -- `S3_ENDPOINT` - S3 endpoint URL (for non-AWS providers) -- `HOLD_PUBLIC` - Allow public reads (default: false) -- `HOLD_OWNER` - DID for captain record creation (optional) -- `HOLD_ALLOW_ALL_CREW` - Allow any authenticated user to register as crew (default: false) +**Scanner config** (env vars only, no YAML/Viper): +- `SCANNER_HOLD_URL`, `SCANNER_SHARED_SECRET`, `SCANNER_WORKERS`, `SCANNER_ADDR` **Credential Helper**: - Token storage: `~/.atcr/credential-helper-token.json` (or Docker's credential store) @@ -706,6 +765,14 @@ See `.env.hold.example` for all available options. Key environment variables: - Storage drivers imported as `_ "github.com/distribution/distribution/v3/registry/storage/driver/s3-aws"` - Hold service reuses distribution's driver factory for multi-backend support +**Configuration system:** +- Config loading uses Viper (`pkg/config/viper.go`) — YAML primary, env vars override +- Config structs use `comment` struct tags for auto-generating commented YAML via `MarshalCommentedYAML()` (`pkg/config/marshal.go`) +- AppView config: `pkg/appview/config.go` (prefix `ATCR_`) +- Hold config: `pkg/hold/config.go` (prefix `HOLD_`, plus standard `AWS_*`/`S3_*` bindings) +- Quota/billing configs are subsections of the hold YAML file, loaded by passing the config path +- Scanner config is env-only (no Viper): `scanner/internal/config/config.go` + **OAuth implementation:** - Client (`pkg/auth/oauth/client.go`) encapsulates all OAuth configuration - Token validation via `com.atproto.server.getSession` ensures no trust in client-provided identity @@ -746,7 +813,7 @@ When writing tests: - Client methods are consistent across authorization, token exchange, and refresh flows **Adding BYOS support for a user**: -1. User sets environment variables (storage credentials, public URL, HOLD_OWNER) +1. User configures hold YAML (storage credentials, public URL, owner DID) 2. User runs hold service - creates captain + crew records in embedded PDS 3. Hold creates `io.atcr.hold.captain` + `io.atcr.hold.crew` records 4. User sets sailor profile `defaultHold` to point to their hold @@ -754,12 +821,9 @@ When writing tests: 6. No AppView changes needed - fully decentralized **Using S3-compatible storage**: -ATCR requires S3-compatible storage. Supported providers: -- AWS S3 - Set `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `S3_BUCKET` -- Storj - Set `S3_ENDPOINT=https://gateway.storjshare.io` -- Minio - Set `S3_ENDPOINT=http://localhost:9000` -- UpCloud - Set `S3_ENDPOINT=https://[bucket-id].upcloudobjects.com` -- Azure/GCS - Use their S3-compatible API endpoints +ATCR requires S3-compatible storage. Configure in hold YAML under `storage:` or via env vars. +Supported providers: AWS S3, Storj (`storage.endpoint: https://gateway.storjshare.io`), +Minio (`storage.endpoint: http://localhost:9000`), UpCloud, Azure/GCS (S3-compatible endpoints). **Working with the database**: - **Base schema** defined in `pkg/appview/db/schema.sql` - source of truth for fresh installations @@ -767,7 +831,7 @@ ATCR requires S3-compatible storage. Supported providers: - **Queries** in `pkg/appview/db/queries.go` - **Stores** for OAuth, devices, sessions in separate files - **Execution order**: schema.sql first, then migrations (automatically on startup) -- **Database path** configurable via `ATCR_UI_DATABASE_PATH` env var +- **Database path** configurable via `ui.database_path` in YAML (or `ATCR_UI_DATABASE_PATH` env var) - **Adding new tables**: Add to `schema.sql` only (no migration needed) - **Altering tables**: Create migration AND update `schema.sql` to keep them in sync diff --git a/pkg/atproto/cbor_gen.go b/pkg/atproto/cbor_gen.go index 755f653..0c425d9 100644 --- a/pkg/atproto/cbor_gen.go +++ b/pkg/atproto/cbor_gen.go @@ -8,6 +8,7 @@ import ( "math" "sort" + util "github.com/bluesky-social/indigo/lex/util" cid "github.com/ipfs/go-cid" cbg "github.com/whyrusleeping/cbor-gen" xerrors "golang.org/x/xerrors" @@ -25,7 +26,7 @@ func (t *CrewRecord) MarshalCBOR(w io.Writer) error { } cw := cbg.NewCborWriter(w) - fieldCount := 6 + fieldCount := 7 if t.Tier == "" { fieldCount-- @@ -153,6 +154,22 @@ func (t *CrewRecord) MarshalCBOR(w io.Writer) error { return err } + // t.Plankowner (bool) (bool) + if len("plankowner") > 8192 { + return xerrors.Errorf("Value in field \"plankowner\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("plankowner"))); err != nil { + return err + } + if _, err := cw.WriteString(string("plankowner")); err != nil { + return err + } + + if err := cbg.WriteBool(w, t.Plankowner); err != nil { + return err + } + // t.Permissions ([]string) (slice) if len("permissions") > 8192 { return xerrors.Errorf("Value in field \"permissions\" was too long") @@ -284,6 +301,24 @@ func (t *CrewRecord) UnmarshalCBOR(r io.Reader) (err error) { t.AddedAt = string(sval) } + // t.Plankowner (bool) (bool) + case "plankowner": + + maj, extra, err = cr.ReadHeader() + if err != nil { + return err + } + if maj != cbg.MajOther { + return fmt.Errorf("booleans must be major type 7") + } + switch extra { + case 20: + t.Plankowner = false + case 21: + t.Plankowner = true + default: + return fmt.Errorf("booleans are either major type 7, value 20 or 21 (got %d)", extra) + } // t.Permissions ([]string) (slice) case "permissions": @@ -1767,3 +1802,549 @@ func (t *StatsRecord) UnmarshalCBOR(r io.Reader) (err error) { return nil } +func (t *ScanRecord) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + + cw := cbg.NewCborWriter(w) + + if _, err := cw.Write([]byte{172}); err != nil { + return err + } + + // t.Low (int64) (int64) + if len("low") > 8192 { + return xerrors.Errorf("Value in field \"low\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("low"))); err != nil { + return err + } + if _, err := cw.WriteString(string("low")); err != nil { + return err + } + + if t.Low >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.Low)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.Low-1)); err != nil { + return err + } + } + + // t.High (int64) (int64) + if len("high") > 8192 { + return xerrors.Errorf("Value in field \"high\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("high"))); err != nil { + return err + } + if _, err := cw.WriteString(string("high")); err != nil { + return err + } + + if t.High >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.High)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.High-1)); err != nil { + return err + } + } + + // t.Type (string) (string) + if len("$type") > 8192 { + return xerrors.Errorf("Value in field \"$type\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("$type"))); err != nil { + return err + } + if _, err := cw.WriteString(string("$type")); err != nil { + return err + } + + if len(t.Type) > 8192 { + return xerrors.Errorf("Value in field t.Type was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Type))); err != nil { + return err + } + if _, err := cw.WriteString(string(t.Type)); err != nil { + return err + } + + // t.Total (int64) (int64) + if len("total") > 8192 { + return xerrors.Errorf("Value in field \"total\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("total"))); err != nil { + return err + } + if _, err := cw.WriteString(string("total")); err != nil { + return err + } + + if t.Total >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.Total)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.Total-1)); err != nil { + return err + } + } + + // t.Medium (int64) (int64) + if len("medium") > 8192 { + return xerrors.Errorf("Value in field \"medium\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("medium"))); err != nil { + return err + } + if _, err := cw.WriteString(string("medium")); err != nil { + return err + } + + if t.Medium >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.Medium)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.Medium-1)); err != nil { + return err + } + } + + // t.UserDID (string) (string) + if len("userDid") > 8192 { + return xerrors.Errorf("Value in field \"userDid\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("userDid"))); err != nil { + return err + } + if _, err := cw.WriteString(string("userDid")); err != nil { + return err + } + + if len(t.UserDID) > 8192 { + return xerrors.Errorf("Value in field t.UserDID was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.UserDID))); err != nil { + return err + } + if _, err := cw.WriteString(string(t.UserDID)); err != nil { + return err + } + + // t.Critical (int64) (int64) + if len("critical") > 8192 { + return xerrors.Errorf("Value in field \"critical\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("critical"))); err != nil { + return err + } + if _, err := cw.WriteString(string("critical")); err != nil { + return err + } + + if t.Critical >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.Critical)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.Critical-1)); err != nil { + return err + } + } + + // t.Manifest (string) (string) + if len("manifest") > 8192 { + return xerrors.Errorf("Value in field \"manifest\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("manifest"))); err != nil { + return err + } + if _, err := cw.WriteString(string("manifest")); err != nil { + return err + } + + if len(t.Manifest) > 8192 { + return xerrors.Errorf("Value in field t.Manifest was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Manifest))); err != nil { + return err + } + if _, err := cw.WriteString(string(t.Manifest)); err != nil { + return err + } + + // t.SbomBlob (util.LexBlob) (struct) + if len("sbomBlob") > 8192 { + return xerrors.Errorf("Value in field \"sbomBlob\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("sbomBlob"))); err != nil { + return err + } + if _, err := cw.WriteString(string("sbomBlob")); err != nil { + return err + } + + if err := t.SbomBlob.MarshalCBOR(cw); err != nil { + return err + } + + // t.ScannedAt (string) (string) + if len("scannedAt") > 8192 { + return xerrors.Errorf("Value in field \"scannedAt\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("scannedAt"))); err != nil { + return err + } + if _, err := cw.WriteString(string("scannedAt")); err != nil { + return err + } + + if len(t.ScannedAt) > 8192 { + return xerrors.Errorf("Value in field t.ScannedAt was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.ScannedAt))); err != nil { + return err + } + if _, err := cw.WriteString(string(t.ScannedAt)); err != nil { + return err + } + + // t.Repository (string) (string) + if len("repository") > 8192 { + return xerrors.Errorf("Value in field \"repository\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("repository"))); err != nil { + return err + } + if _, err := cw.WriteString(string("repository")); err != nil { + return err + } + + if len(t.Repository) > 8192 { + return xerrors.Errorf("Value in field t.Repository was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Repository))); err != nil { + return err + } + if _, err := cw.WriteString(string(t.Repository)); err != nil { + return err + } + + // t.ScannerVersion (string) (string) + if len("scannerVersion") > 8192 { + return xerrors.Errorf("Value in field \"scannerVersion\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("scannerVersion"))); err != nil { + return err + } + if _, err := cw.WriteString(string("scannerVersion")); err != nil { + return err + } + + if len(t.ScannerVersion) > 8192 { + return xerrors.Errorf("Value in field t.ScannerVersion was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.ScannerVersion))); err != nil { + return err + } + if _, err := cw.WriteString(string(t.ScannerVersion)); err != nil { + return err + } + return nil +} + +func (t *ScanRecord) UnmarshalCBOR(r io.Reader) (err error) { + *t = ScanRecord{} + + cr := cbg.NewCborReader(r) + + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + + if maj != cbg.MajMap { + return fmt.Errorf("cbor input should be of type map") + } + + if extra > cbg.MaxLength { + return fmt.Errorf("ScanRecord: map struct too large (%d)", extra) + } + + n := extra + + nameBuf := make([]byte, 14) + for i := uint64(0); i < n; i++ { + nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 8192) + if err != nil { + return err + } + + if !ok { + // Field doesn't exist on this type, so ignore it + if err := cbg.ScanForLinks(cr, func(cid.Cid) {}); err != nil { + return err + } + continue + } + + switch string(nameBuf[:nameLen]) { + // t.Low (int64) (int64) + case "low": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.Low = int64(extraI) + } + // t.High (int64) (int64) + case "high": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.High = int64(extraI) + } + // t.Type (string) (string) + case "$type": + + { + sval, err := cbg.ReadStringWithMax(cr, 8192) + if err != nil { + return err + } + + t.Type = string(sval) + } + // t.Total (int64) (int64) + case "total": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.Total = int64(extraI) + } + // t.Medium (int64) (int64) + case "medium": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.Medium = int64(extraI) + } + // t.UserDID (string) (string) + case "userDid": + + { + sval, err := cbg.ReadStringWithMax(cr, 8192) + if err != nil { + return err + } + + t.UserDID = string(sval) + } + // t.Critical (int64) (int64) + case "critical": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.Critical = int64(extraI) + } + // t.Manifest (string) (string) + case "manifest": + + { + sval, err := cbg.ReadStringWithMax(cr, 8192) + if err != nil { + return err + } + + t.Manifest = string(sval) + } + // t.SbomBlob (util.LexBlob) (struct) + case "sbomBlob": + + { + + b, err := cr.ReadByte() + if err != nil { + return err + } + if b != cbg.CborNull[0] { + if err := cr.UnreadByte(); err != nil { + return err + } + t.SbomBlob = new(util.LexBlob) + if err := t.SbomBlob.UnmarshalCBOR(cr); err != nil { + return xerrors.Errorf("unmarshaling t.SbomBlob pointer: %w", err) + } + } + + } + // t.ScannedAt (string) (string) + case "scannedAt": + + { + sval, err := cbg.ReadStringWithMax(cr, 8192) + if err != nil { + return err + } + + t.ScannedAt = string(sval) + } + // t.Repository (string) (string) + case "repository": + + { + sval, err := cbg.ReadStringWithMax(cr, 8192) + if err != nil { + return err + } + + t.Repository = string(sval) + } + // t.ScannerVersion (string) (string) + case "scannerVersion": + + { + sval, err := cbg.ReadStringWithMax(cr, 8192) + if err != nil { + return err + } + + t.ScannerVersion = string(sval) + } + + default: + // Field doesn't exist on this type, so ignore it + if err := cbg.ScanForLinks(r, func(cid.Cid) {}); err != nil { + return err + } + } + } + + return nil +} diff --git a/pkg/atproto/generate.go b/pkg/atproto/generate.go index 626a1f1..f8dda39 100644 --- a/pkg/atproto/generate.go +++ b/pkg/atproto/generate.go @@ -32,6 +32,7 @@ func main() { atproto.LayerRecord{}, atproto.TangledProfileRecord{}, atproto.StatsRecord{}, + atproto.ScanRecord{}, ); err != nil { fmt.Printf("Failed to generate CBOR encoders: %v\n", err) os.Exit(1) diff --git a/pkg/atproto/lexicon.go b/pkg/atproto/lexicon.go index d5df353..b1812af 100644 --- a/pkg/atproto/lexicon.go +++ b/pkg/atproto/lexicon.go @@ -10,6 +10,8 @@ import ( "fmt" "strings" "time" + + lexutil "github.com/bluesky-social/indigo/lex/util" ) // Collection names for ATProto records @@ -41,6 +43,10 @@ const ( // Stored in hold's embedded PDS to track pull/push counts per owner+repo StatsCollection = "io.atcr.hold.stats" + // ScanCollection is the collection name for vulnerability scan results + // Stored in hold's embedded PDS to track scan results per manifest + ScanCollection = "io.atcr.hold.scan" + // TangledProfileCollection is the collection name for tangled profiles // Stored in hold's embedded PDS (singleton record at rkey "self") TangledProfileCollection = "sh.tangled.actor.profile" @@ -594,7 +600,7 @@ type CrewRecord struct { Role string `json:"role" cborgen:"role"` Permissions []string `json:"permissions" cborgen:"permissions"` Tier string `json:"tier,omitempty" cborgen:"tier,omitempty"` // Optional tier for quota limits (e.g., 'deckhand', 'bosun', 'quartermaster') - Plankowner bool `json:"plankowner,omitempty" cborgen:"plankowner,omitempty"` // Early adopter flag - gets plankowner_crew_tier for free + Plankowner bool `json:"plankowner,omitempty" cborgen:"plankowner"` // Early adopter flag - gets plankowner_crew_tier for free AddedAt string `json:"addedAt" cborgen:"addedAt"` // RFC3339 timestamp } @@ -675,6 +681,55 @@ func CrewRecordKey(memberDID string) string { return strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(hash[:16])) } +// ScanRecord represents vulnerability scan results for a manifest +// Collection: io.atcr.hold.scan +// Stored in hold's embedded PDS to track scan results per manifest +// Uses CBOR encoding for efficient storage in hold's carstore +// RKey is deterministic: based on manifest digest (one scan per manifest) +type ScanRecord struct { + Type string `json:"$type" cborgen:"$type"` + Manifest string `json:"manifest" cborgen:"manifest"` // AT-URI of the scanned manifest (e.g., "at://did:plc:xyz/io.atcr.manifest/abc123...") + Repository string `json:"repository" cborgen:"repository"` // Repository name (e.g., "myapp") + UserDID string `json:"userDid" cborgen:"userDid"` // DID of the image owner + SbomBlob *lexutil.LexBlob `json:"sbomBlob,omitempty" cborgen:"sbomBlob"` // SBOM blob uploaded to hold's PDS blob storage + Critical int64 `json:"critical" cborgen:"critical"` // Count of critical vulnerabilities + High int64 `json:"high" cborgen:"high"` // Count of high vulnerabilities + Medium int64 `json:"medium" cborgen:"medium"` // Count of medium vulnerabilities + Low int64 `json:"low" cborgen:"low"` // Count of low vulnerabilities + Total int64 `json:"total" cborgen:"total"` // Total vulnerability count + ScannerVersion string `json:"scannerVersion" cborgen:"scannerVersion"` // Scanner version (e.g., "atcr-scanner-v1.0.0") + ScannedAt string `json:"scannedAt" cborgen:"scannedAt"` // RFC3339 timestamp of scan completion +} + +// NewScanRecord creates a new scan record +// manifestDigest: the manifest digest (e.g., "sha256:abc123...") +// userDID: the DID of the image owner (used to build the manifest AT-URI) +// sbomBlob: blob reference from uploading SBOM to PDS blob storage (nil if no SBOM) +func NewScanRecord(manifestDigest, repository, userDID string, sbomBlob *lexutil.LexBlob, critical, high, medium, low, total int, scannerVersion string) *ScanRecord { + return &ScanRecord{ + Type: ScanCollection, + Manifest: BuildManifestURI(userDID, manifestDigest), + Repository: repository, + UserDID: userDID, + SbomBlob: sbomBlob, + Critical: int64(critical), + High: int64(high), + Medium: int64(medium), + Low: int64(low), + Total: int64(total), + ScannerVersion: scannerVersion, + ScannedAt: time.Now().Format(time.RFC3339), + } +} + +// ScanRecordKey generates a deterministic record key for a scan result +// Uses the manifest digest (without algorithm prefix) as the rkey +// This ensures one scan record per manifest, and re-scans upsert the record +func ScanRecordKey(manifestDigest string) string { + // Remove the "sha256:" prefix - the hex digest is already a valid rkey + return strings.TrimPrefix(manifestDigest, "sha256:") +} + // TangledProfileRecord represents a Tangled profile for the hold // Collection: sh.tangled.actor.profile (singleton record at rkey "self") // Stored in the hold's embedded PDS diff --git a/pkg/hold/pds/scan.go b/pkg/hold/pds/scan.go new file mode 100644 index 0000000..77c464f --- /dev/null +++ b/pkg/hold/pds/scan.go @@ -0,0 +1,59 @@ +package pds + +import ( + "context" + "fmt" + + "atcr.io/pkg/atproto" + "github.com/ipfs/go-cid" +) + +// CreateScanRecord creates or updates a scan result record in the hold's PDS +// Uses a deterministic rkey based on the manifest digest, so re-scans upsert +func (p *HoldPDS) CreateScanRecord(ctx context.Context, record *atproto.ScanRecord) (string, cid.Cid, error) { + if record.Type != atproto.ScanCollection { + return "", cid.Undef, fmt.Errorf("invalid record type: %s", record.Type) + } + + if record.Manifest == "" { + return "", cid.Undef, fmt.Errorf("manifest AT-URI is required") + } + + // Extract the digest from the manifest AT-URI to use as rkey + manifestDigest, err := atproto.ParseManifestURI(record.Manifest) + if err != nil { + return "", cid.Undef, fmt.Errorf("invalid manifest AT-URI: %w", err) + } + rkey := atproto.ScanRecordKey(manifestDigest) + + // Upsert: re-scans update the existing record + rpath, recordCID, _, err := p.repomgr.UpsertRecord( + ctx, + p.uid, + atproto.ScanCollection, + rkey, + record, + ) + if err != nil { + return "", cid.Undef, fmt.Errorf("failed to upsert scan record: %w", err) + } + + return rpath, recordCID, nil +} + +// GetScanRecord retrieves a scan result record by manifest digest +func (p *HoldPDS) GetScanRecord(ctx context.Context, manifestDigest string) (cid.Cid, *atproto.ScanRecord, error) { + rkey := atproto.ScanRecordKey(manifestDigest) + + recordCID, val, err := p.repomgr.GetRecord(ctx, p.uid, atproto.ScanCollection, rkey, cid.Undef) + if err != nil { + return cid.Undef, nil, fmt.Errorf("failed to get scan record: %w", err) + } + + scanRecord, ok := val.(*atproto.ScanRecord) + if !ok { + return cid.Undef, nil, fmt.Errorf("unexpected type for scan record: %T", val) + } + + return recordCID, scanRecord, nil +} diff --git a/pkg/hold/pds/scan_broadcaster.go b/pkg/hold/pds/scan_broadcaster.go index 405060a..c527768 100644 --- a/pkg/hold/pds/scan_broadcaster.go +++ b/pkg/hold/pds/scan_broadcaster.go @@ -3,7 +3,6 @@ package pds import ( "context" "crypto/rand" - "crypto/sha256" "database/sql" "encoding/hex" "encoding/json" @@ -12,6 +11,8 @@ import ( "sync" "time" + "atcr.io/pkg/atproto" + lexutil "github.com/bluesky-social/indigo/lex/util" storagedriver "github.com/distribution/distribution/v3/registry/storage/driver" "github.com/gorilla/websocket" ) @@ -353,7 +354,7 @@ func (sb *ScanBroadcaster) handleAck(sub *ScanSubscriber, seq int64) { "subscriberId", sub.id) } -// handleResult processes a completed scan result: stores ORAS manifest + marks completed +// handleResult processes a completed scan result: uploads SBOM blob + stores scan record in PDS func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) { ctx := context.Background() @@ -382,36 +383,40 @@ func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) return } - // Store vulnerability report blob in S3 - if msg.VulnReport != "" { - vulnJSON := []byte(msg.VulnReport) - vulnDigest := fmt.Sprintf("sha256:%x", sha256.Sum256(vulnJSON)) - - if err := sb.uploadBlob(ctx, vulnDigest, vulnJSON); err != nil { - slog.Error("Failed to upload vulnerability report blob", + // Upload SBOM as a blob to the hold's PDS blob storage (like manifest blobs) + var sbomBlob *lexutil.LexBlob + if msg.SBOM != "" { + blob, err := uploadBlobToStorage(ctx, sb.driver, sb.holdDID, []byte(msg.SBOM), "application/spdx+json") + if err != nil { + slog.Error("Failed to upload SBOM blob to PDS storage", "seq", msg.Seq, "error", err) - } - - // Build and store ORAS manifest - if msg.Summary != nil { - if err := sb.storeORASManifest(ctx, manifestDigest, repository, userDID, vulnDigest, vulnJSON, *msg.Summary); err != nil { - slog.Error("Failed to store ORAS manifest", - "seq", msg.Seq, - "error", err) - } + } else { + sbomBlob = blob } } - // Store SBOM blob if provided - if msg.SBOM != "" { - sbomJSON := []byte(msg.SBOM) - sbomDigest := fmt.Sprintf("sha256:%x", sha256.Sum256(sbomJSON)) + // Store scan result as a record in the hold's embedded PDS + if msg.Summary != nil { + scanRecord := atproto.NewScanRecord( + manifestDigest, repository, userDID, + sbomBlob, + msg.Summary.Critical, msg.Summary.High, msg.Summary.Medium, msg.Summary.Low, msg.Summary.Total, + "atcr-scanner-v1.0.0", + ) - if err := sb.uploadBlob(ctx, sbomDigest, sbomJSON); err != nil { - slog.Error("Failed to upload SBOM blob", + rpath, _, err := sb.pds.CreateScanRecord(ctx, scanRecord) + if err != nil { + slog.Error("Failed to store scan record in PDS", "seq", msg.Seq, "error", err) + } else { + slog.Info("Scan record stored in PDS", + "rpath", rpath, + "manifest", scanRecord.Manifest, + "critical", msg.Summary.Critical, + "high", msg.Summary.High, + "total", msg.Summary.Total) } } @@ -584,99 +589,6 @@ func (sb *ScanBroadcaster) ValidateScannerSecret(secret string) bool { return sb.secret != "" && secret == sb.secret } -// storeORASManifest creates an ORAS vulnerability manifest as a blob in S3 -// The ORAS manifest's "subject" field references the original manifest by digest, -// enabling OCI referrers API discovery. -func (sb *ScanBroadcaster) storeORASManifest(ctx context.Context, manifestDigest, repository, userDID, vulnDigest string, vulnJSON []byte, summary VulnerabilitySummary) error { - scannerVersion := "atcr-scanner-v1.0.0" - - // Create ORAS manifest - orasManifest := map[string]interface{}{ - "schemaVersion": 2, - "mediaType": "application/vnd.oci.image.manifest.v1+json", - "artifactType": "application/vnd.atcr.vulnerabilities+json", - "config": map[string]interface{}{ - "mediaType": "application/vnd.oci.empty.v1+json", - "digest": "sha256:44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a", - "size": 2, - }, - "subject": map[string]interface{}{ - "mediaType": "application/vnd.oci.image.manifest.v1+json", - "digest": manifestDigest, - "size": 0, - }, - "layers": []map[string]interface{}{ - { - "mediaType": "application/json", - "digest": vulnDigest, - "size": len(vulnJSON), - "annotations": map[string]string{ - "org.opencontainers.image.title": "vulnerability-report.json", - }, - }, - }, - "annotations": map[string]string{ - "io.atcr.vuln.critical": fmt.Sprintf("%d", summary.Critical), - "io.atcr.vuln.high": fmt.Sprintf("%d", summary.High), - "io.atcr.vuln.medium": fmt.Sprintf("%d", summary.Medium), - "io.atcr.vuln.low": fmt.Sprintf("%d", summary.Low), - "io.atcr.vuln.total": fmt.Sprintf("%d", summary.Total), - "io.atcr.vuln.scannedAt": time.Now().Format(time.RFC3339), - "io.atcr.vuln.scannerVersion": scannerVersion, - "io.atcr.vuln.repository": repository, - "io.atcr.vuln.ownerDid": userDID, - "io.atcr.vuln.holdDid": sb.holdDID, - }, - } - - orasManifestJSON, err := json.Marshal(orasManifest) - if err != nil { - return fmt.Errorf("failed to encode ORAS manifest: %w", err) - } - - orasHash := sha256.Sum256(orasManifestJSON) - orasDigest := fmt.Sprintf("sha256:%x", orasHash) - - // Upload ORAS manifest blob to S3 - if err := sb.uploadBlob(ctx, orasDigest, orasManifestJSON); err != nil { - return fmt.Errorf("failed to upload ORAS manifest blob: %w", err) - } - - slog.Info("ORAS manifest stored", - "digest", orasDigest, - "repository", repository, - "userDid", userDID, - "critical", summary.Critical, - "high", summary.High, - "total", summary.Total) - - return nil -} - -// uploadBlob uploads a blob to S3 storage -func (sb *ScanBroadcaster) uploadBlob(ctx context.Context, digest string, data []byte) error { - digestHex := digest[len("sha256:"):] - if len(digestHex) < 2 { - return fmt.Errorf("invalid digest: %s", digest) - } - - blobPath := fmt.Sprintf("/docker/registry/v2/blobs/sha256/%s/%s/data", - digestHex[:2], digestHex) - - writer, err := sb.driver.Writer(ctx, blobPath, false) - if err != nil { - return fmt.Errorf("failed to create storage writer: %w", err) - } - defer writer.Close() - - if _, err := writer.Write(data); err != nil { - writer.Cancel(ctx) - return fmt.Errorf("failed to write blob data: %w", err) - } - - return writer.Commit(ctx) -} - func generateSubscriberID() string { b := make([]byte, 8) rand.Read(b) diff --git a/pkg/hold/pds/server.go b/pkg/hold/pds/server.go index 31414d7..c6a895f 100644 --- a/pkg/hold/pds/server.go +++ b/pkg/hold/pds/server.go @@ -29,6 +29,7 @@ func init() { lexutil.RegisterType(atproto.LayerCollection, &atproto.LayerRecord{}) lexutil.RegisterType(atproto.TangledProfileCollection, &atproto.TangledProfileRecord{}) lexutil.RegisterType(atproto.StatsCollection, &atproto.StatsRecord{}) + lexutil.RegisterType(atproto.ScanCollection, &atproto.ScanRecord{}) } // HoldPDS is a minimal ATProto PDS implementation for a hold service