diff --git a/Dockerfile.hold b/Dockerfile.hold index a2ac8b9..32f5734 100644 --- a/Dockerfile.hold +++ b/Dockerfile.hold @@ -19,7 +19,7 @@ COPY . . # Build frontend assets (Tailwind CSS, JS bundle, SVG icons) RUN npm ci -RUN npm run css:build && npm run css:copy-hold && npm run js:build:hold && npm run icons:build +RUN go generate ./... # Conditionally add billing tag based on build arg RUN if [ "$BILLING_ENABLED" = "true" ]; then \ diff --git a/pkg/appview/config.go b/pkg/appview/config.go index 9ae8322..898b574 100644 --- a/pkg/appview/config.go +++ b/pkg/appview/config.go @@ -312,7 +312,6 @@ func buildDistributionConfig(cfg *Config, v *viper.Viper) (*configuration.Config "service": cfg.Auth.ServiceName, "issuer": cfg.Auth.ServiceName, "rootcertbundle": cfg.Auth.CertPath, - "privatekey": cfg.Auth.KeyPath, "expiration": int(cfg.Auth.TokenExpiration.Seconds()), }, } diff --git a/pkg/appview/crypto_keys.go b/pkg/appview/crypto_keys.go new file mode 100644 index 0000000..c7ec79c --- /dev/null +++ b/pkg/appview/crypto_keys.go @@ -0,0 +1,182 @@ +package appview + +import ( + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "crypto/x509/pkix" + "database/sql" + "encoding/pem" + "fmt" + "log/slog" + "math/big" + "os" + "path/filepath" + "time" + + "atcr.io/pkg/appview/db" + "github.com/bluesky-social/indigo/atproto/atcrypto" +) + +// loadOAuthKey loads the OAuth P-256 key with priority: DB → file → generate. +// Keys loaded from file or newly generated are stored in the DB. +func loadOAuthKey(database *sql.DB, keyPath string) (*atcrypto.PrivateKeyP256, error) { + // Try database first + data, err := db.GetCryptoKey(database, "oauth_p256") + if err != nil { + return nil, fmt.Errorf("failed to query crypto_keys: %w", err) + } + if data != nil { + key, err := atcrypto.ParsePrivateBytesP256(data) + if err != nil { + return nil, fmt.Errorf("failed to parse OAuth key from database: %w", err) + } + slog.Info("Loaded OAuth P-256 key from database") + return key, nil + } + + // Try file fallback + if keyPath != "" { + if fileData, err := os.ReadFile(keyPath); err == nil { + key, err := atcrypto.ParsePrivateBytesP256(fileData) + if err != nil { + return nil, fmt.Errorf("failed to parse OAuth key from file %s: %w", keyPath, err) + } + // Migrate to database + if err := db.PutCryptoKey(database, "oauth_p256", fileData); err != nil { + return nil, fmt.Errorf("failed to store OAuth key in database: %w", err) + } + slog.Info("Migrated OAuth P-256 key from file to database", "path", keyPath) + return key, nil + } + } + + // Generate new key + p256Key, err := atcrypto.GeneratePrivateKeyP256() + if err != nil { + return nil, fmt.Errorf("failed to generate OAuth P-256 key: %w", err) + } + + keyBytes := p256Key.Bytes() + if err := db.PutCryptoKey(database, "oauth_p256", keyBytes); err != nil { + return nil, fmt.Errorf("failed to store generated OAuth key in database: %w", err) + } + slog.Info("Generated new OAuth P-256 key and stored in database") + + return p256Key, nil +} + +// loadJWTKeyAndCert loads the JWT RSA key from DB (with file fallback) and generates +// a self-signed certificate. The cert is always regenerated and written to certPath +// on disk because the distribution library reads it via os.Open(). +func loadJWTKeyAndCert(database *sql.DB, keyPath, certPath string) (*rsa.PrivateKey, []byte, error) { + rsaKey, err := loadRSAKey(database, keyPath) + if err != nil { + return nil, nil, err + } + + // Generate cert and write to disk for distribution library + certDER, err := generateAndWriteCert(rsaKey, certPath) + if err != nil { + return nil, nil, err + } + + return rsaKey, certDER, nil +} + +// loadRSAKey loads the RSA private key with priority: DB → file → generate. +func loadRSAKey(database *sql.DB, keyPath string) (*rsa.PrivateKey, error) { + // Try database first + data, err := db.GetCryptoKey(database, "jwt_rsa") + if err != nil { + return nil, fmt.Errorf("failed to query crypto_keys: %w", err) + } + if data != nil { + key, err := parseRSAKeyPEM(data) + if err != nil { + return nil, fmt.Errorf("failed to parse RSA key from database: %w", err) + } + slog.Info("Loaded JWT RSA key from database") + return key, nil + } + + // Try file fallback + if keyPath != "" { + if fileData, err := os.ReadFile(keyPath); err == nil { + key, err := parseRSAKeyPEM(fileData) + if err != nil { + return nil, fmt.Errorf("failed to parse RSA key from file %s: %w", keyPath, err) + } + // Migrate to database + if err := db.PutCryptoKey(database, "jwt_rsa", fileData); err != nil { + return nil, fmt.Errorf("failed to store RSA key in database: %w", err) + } + slog.Info("Migrated JWT RSA key from file to database", "path", keyPath) + return key, nil + } + } + + // Generate new key + rsaKey, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + return nil, fmt.Errorf("failed to generate RSA key: %w", err) + } + + keyPEM := pem.EncodeToMemory(&pem.Block{ + Type: "RSA PRIVATE KEY", + Bytes: x509.MarshalPKCS1PrivateKey(rsaKey), + }) + if err := db.PutCryptoKey(database, "jwt_rsa", keyPEM); err != nil { + return nil, fmt.Errorf("failed to store generated RSA key in database: %w", err) + } + slog.Info("Generated new JWT RSA key and stored in database") + + return rsaKey, nil +} + +func parseRSAKeyPEM(data []byte) (*rsa.PrivateKey, error) { + block, _ := pem.Decode(data) + if block == nil || block.Type != "RSA PRIVATE KEY" { + return nil, fmt.Errorf("failed to decode PEM block containing RSA private key") + } + return x509.ParsePKCS1PrivateKey(block.Bytes) +} + +// generateAndWriteCert creates a self-signed certificate from the RSA key and writes +// it to certPath. Returns the DER-encoded certificate bytes for the JWT x5c header. +func generateAndWriteCert(rsaKey *rsa.PrivateKey, certPath string) ([]byte, error) { + template := x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{ + Organization: []string{"ATCR"}, + CommonName: "ATCR Token Signing Certificate", + }, + NotBefore: time.Now(), + NotAfter: time.Now().Add(10 * 365 * 24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + BasicConstraintsValid: true, + } + + certDER, err := x509.CreateCertificate(rand.Reader, &template, &template, &rsaKey.PublicKey, rsaKey) + if err != nil { + return nil, fmt.Errorf("failed to create certificate: %w", err) + } + + // Write cert to disk for distribution library + certPEM := pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", + Bytes: certDER, + }) + + dir := filepath.Dir(certPath) + if err := os.MkdirAll(dir, 0700); err != nil { + return nil, fmt.Errorf("failed to create cert directory: %w", err) + } + if err := os.WriteFile(certPath, certPEM, 0644); err != nil { + return nil, fmt.Errorf("failed to write certificate: %w", err) + } + + slog.Info("Generated JWT signing certificate", "path", certPath) + return certDER, nil +} diff --git a/pkg/appview/db/crypto_keys.go b/pkg/appview/db/crypto_keys.go new file mode 100644 index 0000000..e6338b4 --- /dev/null +++ b/pkg/appview/db/crypto_keys.go @@ -0,0 +1,26 @@ +package db + +import "database/sql" + +// GetCryptoKey retrieves a key by name from the database. +// Returns nil, nil if no key with that name exists. +func GetCryptoKey(db DBTX, name string) ([]byte, error) { + var data []byte + err := db.QueryRow("SELECT key_data FROM crypto_keys WHERE name = ?", name).Scan(&data) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, err + } + return data, nil +} + +// PutCryptoKey stores a key in the database, replacing any existing key with the same name. +func PutCryptoKey(db DBTX, name string, data []byte) error { + _, err := db.Exec( + "INSERT INTO crypto_keys (name, key_data) VALUES (?, ?) ON CONFLICT(name) DO UPDATE SET key_data = excluded.key_data", + name, data, + ) + return err +} diff --git a/pkg/appview/db/migrations/0013_create_crypto_keys.yaml b/pkg/appview/db/migrations/0013_create_crypto_keys.yaml new file mode 100644 index 0000000..f85ef92 --- /dev/null +++ b/pkg/appview/db/migrations/0013_create_crypto_keys.yaml @@ -0,0 +1,7 @@ +description: Create crypto_keys table for storing signing keys in the database +query: | + CREATE TABLE IF NOT EXISTS crypto_keys ( + name TEXT PRIMARY KEY, + key_data BLOB NOT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + ); diff --git a/pkg/appview/db/schema.sql b/pkg/appview/db/schema.sql index 7b59bf4..c303697 100644 --- a/pkg/appview/db/schema.sql +++ b/pkg/appview/db/schema.sql @@ -237,3 +237,9 @@ CREATE TABLE IF NOT EXISTS repo_pages ( FOREIGN KEY(did) REFERENCES users(did) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_repo_pages_did ON repo_pages(did); + +CREATE TABLE IF NOT EXISTS crypto_keys ( + name TEXT PRIMARY KEY, + key_data BLOB NOT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP +); diff --git a/pkg/appview/handlers/scan_result_test.go b/pkg/appview/handlers/scan_result_test.go index 5e52f29..96a33b1 100644 --- a/pkg/appview/handlers/scan_result_test.go +++ b/pkg/appview/handlers/scan_result_test.go @@ -84,24 +84,24 @@ func TestScanResult_WithVulnerabilities(t *testing.T) { body := rr.Body.String() - // Should contain severity badges - if !strings.Contains(body, "badge-error") { - t.Error("Expected body to contain badge-error for critical vulnerabilities") + // Should contain vuln-strip severity boxes + if !strings.Contains(body, "vuln-box-critical") { + t.Error("Expected body to contain vuln-box-critical for critical vulnerabilities") } - if !strings.Contains(body, "C:2") { - t.Error("Expected body to contain 'C:2' for critical count") + if !strings.Contains(body, `data-tip="Critical">2<`) { + t.Error("Expected critical count of 2") } - if !strings.Contains(body, "badge-warning") { - t.Error("Expected body to contain badge-warning for high vulnerabilities") + if !strings.Contains(body, "vuln-box-high") { + t.Error("Expected body to contain vuln-box-high for high vulnerabilities") } - if !strings.Contains(body, "H:5") { - t.Error("Expected body to contain 'H:5' for high count") + if !strings.Contains(body, `data-tip="High">5<`) { + t.Error("Expected high count of 5") } - if !strings.Contains(body, "M:10") { - t.Error("Expected body to contain 'M:10' for medium count") + if !strings.Contains(body, `data-tip="Medium">10<`) { + t.Error("Expected medium count of 10") } - if !strings.Contains(body, "L:3") { - t.Error("Expected body to contain 'L:3' for low count") + if !strings.Contains(body, `data-tip="Low">3<`) { + t.Error("Expected low count of 3") } // Should be clickable (has openVulnDetails) if !strings.Contains(body, "openVulnDetails") { @@ -267,8 +267,8 @@ func TestScanResult_OnlyCriticalShown(t *testing.T) { body := rr.Body.String() - if !strings.Contains(body, "C:3") { - t.Error("Expected body to contain 'C:3'") + if !strings.Contains(body, `data-tip="Critical">3<`) { + t.Error("Expected critical count of 3") } // Zero-count badges should NOT appear if strings.Contains(body, "H:0") { @@ -346,8 +346,8 @@ func TestBatchScanResult_MultipleDigests(t *testing.T) { } // abc123 should have vulnerability badges - if !strings.Contains(body, "C:2") { - t.Error("Expected body to contain 'C:2' for abc123") + if !strings.Contains(body, `data-tip="Critical">2<`) { + t.Error("Expected critical count of 2 for abc123") } // def456 should have clean badge if !strings.Contains(body, "Clean") { @@ -430,7 +430,7 @@ func TestBatchScanResult_SingleDigest(t *testing.T) { if !strings.Contains(body, `id="scan-badge-abc123"`) { t.Error("Expected OOB span for abc123") } - if !strings.Contains(body, "C:1") { - t.Error("Expected body to contain 'C:1'") + if !strings.Contains(body, `data-tip="Critical">1<`) { + t.Error("Expected critical count of 1") } } diff --git a/pkg/appview/handlers/vuln_details_test.go b/pkg/appview/handlers/vuln_details_test.go index 80c1bdd..7520880 100644 --- a/pkg/appview/handlers/vuln_details_test.go +++ b/pkg/appview/handlers/vuln_details_test.go @@ -17,42 +17,43 @@ import ( // mockGrypeReport returns a minimal Grype JSON report func mockGrypeReport() string { + // Grype v0.107+ uses PascalCase JSON keys, and severity is nested under Metadata report := map[string]any{ "matches": []map[string]any{ { - "vulnerability": map[string]any{ - "id": "CVE-2024-1234", - "severity": "Critical", - "fix": map[string]any{"versions": []string{"1.2.4"}, "state": "fixed"}, + "Vulnerability": map[string]any{ + "ID": "CVE-2024-1234", + "Metadata": map[string]any{"Severity": "Critical"}, + "Fix": map[string]any{"Versions": []string{"1.2.4"}, "State": "fixed"}, }, - "artifact": map[string]any{ - "name": "libssl", - "version": "1.1.1", - "type": "deb", + "Package": map[string]any{ + "Name": "libssl", + "Version": "1.1.1", + "Type": "deb", }, }, { - "vulnerability": map[string]any{ - "id": "CVE-2024-5678", - "severity": "Low", - "fix": map[string]any{"versions": []string{}, "state": "not-fixed"}, + "Vulnerability": map[string]any{ + "ID": "CVE-2024-5678", + "Metadata": map[string]any{"Severity": "Low"}, + "Fix": map[string]any{"Versions": []string{}, "State": "not-fixed"}, }, - "artifact": map[string]any{ - "name": "zlib", - "version": "1.2.11", - "type": "deb", + "Package": map[string]any{ + "Name": "zlib", + "Version": "1.2.11", + "Type": "deb", }, }, { - "vulnerability": map[string]any{ - "id": "GHSA-abcd-efgh-ijkl", - "severity": "High", - "fix": map[string]any{"versions": []string{"2.0.0"}, "state": "fixed"}, + "Vulnerability": map[string]any{ + "ID": "GHSA-abcd-efgh-ijkl", + "Metadata": map[string]any{"Severity": "High"}, + "Fix": map[string]any{"Versions": []string{"2.0.0"}, "State": "fixed"}, }, - "artifact": map[string]any{ - "name": "express", - "version": "4.17.1", - "type": "npm", + "Package": map[string]any{ + "Name": "express", + "Version": "4.17.1", + "Type": "npm", }, }, }, @@ -251,9 +252,12 @@ func TestVulnDetails_NoVulnReportBlob(t *testing.T) { body := rr.Body.String() - // Should show summary counts - if !strings.Contains(body, "2 Critical") { - t.Error("Expected body to contain '2 Critical' summary") + // Should show summary counts in vuln-strip boxes + if !strings.Contains(body, "vuln-box-critical") { + t.Error("Expected body to contain vuln-box-critical in summary") + } + if !strings.Contains(body, `data-tip="Critical">2<`) { + t.Error("Expected critical count of 2 in summary") } // Should indicate no detailed report diff --git a/pkg/appview/server.go b/pkg/appview/server.go index 1160d67..293d3a6 100644 --- a/pkg/appview/server.go +++ b/pkg/appview/server.go @@ -185,10 +185,15 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer, slog.Info("TEST_MODE enabled - will use HTTP for local DID resolution") } + // Load crypto keys from database (with file fallback and migration) + oauthKey, err := loadOAuthKey(s.Database, cfg.Server.OAuthKeyPath) + if err != nil { + return nil, fmt.Errorf("failed to load OAuth key: %w", err) + } + // Create OAuth client app desiredScopes := oauth.GetDefaultScopes(defaultHoldDID) - var err error - s.OAuthClientApp, err = oauth.NewClientApp(baseURL, s.OAuthStore, desiredScopes, cfg.Server.OAuthKeyPath, cfg.Server.ClientName) + s.OAuthClientApp, err = oauth.NewClientAppWithKey(baseURL, s.OAuthStore, desiredScopes, oauthKey, cfg.Server.ClientName) if err != nil { return nil, fmt.Errorf("failed to create OAuth client app: %w", err) } @@ -404,11 +409,12 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer, // Create token issuer if cfg.Distribution.Auth["token"] != nil { - s.TokenIssuer, err = s.createTokenIssuer() + rsaKey, certDER, err := loadJWTKeyAndCert(s.Database, cfg.Auth.KeyPath, cfg.Auth.CertPath) if err != nil { - return nil, fmt.Errorf("failed to create token issuer: %w", err) + return nil, fmt.Errorf("failed to load JWT key material: %w", err) } - slog.Info("Auth keys initialized", "path", cfg.Auth.KeyPath) + s.TokenIssuer = token.NewIssuerFromKey(rsaKey, certDER, cfg.Auth.ServiceName, cfg.Auth.ServiceName, cfg.Auth.TokenExpiration) + slog.Info("Auth keys initialized") } // Create registry app (distribution library handler) @@ -593,15 +599,6 @@ func (s *AppViewServer) Serve() error { return nil } -// createTokenIssuer creates a token issuer for auth handlers. -func (s *AppViewServer) createTokenIssuer() (*token.Issuer, error) { - return token.NewIssuer( - s.Config.Auth.KeyPath, - s.Config.Auth.ServiceName, - s.Config.Auth.ServiceName, - s.Config.Auth.TokenExpiration, - ) -} // DomainRoutingMiddleware enforces three-tier domain routing: // diff --git a/pkg/auth/oauth/client.go b/pkg/auth/oauth/client.go index d6aab30..ce97085 100644 --- a/pkg/auth/oauth/client.go +++ b/pkg/auth/oauth/client.go @@ -13,6 +13,7 @@ import ( "time" "atcr.io/pkg/atproto" + "github.com/bluesky-social/indigo/atproto/atcrypto" "github.com/bluesky-social/indigo/atproto/auth/oauth" "github.com/bluesky-social/indigo/atproto/syntax" ) @@ -95,6 +96,38 @@ func NewClientApp(baseURL string, store oauth.ClientAuthStore, scopes []string, return clientApp, nil } +// NewClientAppWithKey creates an indigo OAuth ClientApp with a pre-loaded P-256 key. +// Used by AppView when loading keys from the database instead of disk. +// For localhost development, privateKey is ignored (public client). +func NewClientAppWithKey(baseURL string, store oauth.ClientAuthStore, scopes []string, privateKey *atcrypto.PrivateKeyP256, clientName string) (*oauth.ClientApp, error) { + var config oauth.ClientConfig + redirectURI := RedirectURI(baseURL) + + if !isLocalhost(baseURL) { + clientID := baseURL + "/oauth-client-metadata.json" + config = oauth.NewPublicConfig(clientID, redirectURI, scopes) + + keyID, err := GenerateKeyID(privateKey) + if err != nil { + return nil, fmt.Errorf("failed to generate key ID: %w", err) + } + + if err := config.SetClientSecret(privateKey, keyID); err != nil { + return nil, fmt.Errorf("failed to configure confidential client: %w", err) + } + + slog.Info("Configured confidential OAuth client", "key_id", keyID) + } else { + config = oauth.NewLocalhostConfig(redirectURI, scopes) + slog.Info("Using public OAuth client (localhost development)") + } + + clientApp := oauth.NewClientApp(&config, store) + clientApp.Dir = atproto.GetDirectory() + + return clientApp, nil +} + // RedirectURI returns the OAuth redirect URI for ATCR func RedirectURI(baseURL string) string { return baseURL + "/auth/oauth/callback" diff --git a/pkg/auth/token/issuer.go b/pkg/auth/token/issuer.go index ea126ec..08694a9 100644 --- a/pkg/auth/token/issuer.go +++ b/pkg/auth/token/issuer.go @@ -59,6 +59,19 @@ func NewIssuer(privateKeyPath, issuer, service string, expiration time.Duration) }, nil } +// NewIssuerFromKey creates a JWT issuer from pre-loaded key material. +// certDER is the DER-encoded X.509 certificate for the x5c JWT header. +func NewIssuerFromKey(privateKey *rsa.PrivateKey, certDER []byte, issuer, service string, expiration time.Duration) *Issuer { + return &Issuer{ + privateKey: privateKey, + publicKey: &privateKey.PublicKey, + certificate: certDER, + issuer: issuer, + service: service, + expiration: expiration, + } +} + // Issue creates and signs a new JWT token func (i *Issuer) Issue(subject string, access []auth.AccessEntry, authMethod string) (string, error) { claims := NewClaims(subject, i.issuer, i.service, i.expiration, access, authMethod) diff --git a/pkg/hold/config.go b/pkg/hold/config.go index 04eaf33..13f06c5 100644 --- a/pkg/hold/config.go +++ b/pkg/hold/config.go @@ -139,6 +139,9 @@ type ServerConfig struct { type ScannerConfig struct { // Shared secret for scanner WebSocket authentication. Empty disables scanning. Secret string `yaml:"secret" comment:"Shared secret for scanner WebSocket auth. Empty disables scanning."` + + // Minimum interval between re-scans of the same manifest. 0 disables proactive scanning. + RescanInterval time.Duration `yaml:"rescan_interval" comment:"Minimum interval between re-scans of the same manifest. When set, the hold proactively scans manifests when the scanner is idle. Default: 24h. Set to 0 to disable."` } // DatabaseConfig defines embedded PDS database settings @@ -220,6 +223,7 @@ func setHoldDefaults(v *viper.Viper) { v.SetDefault("gc.enabled", false) // Scanner defaults v.SetDefault("scanner.secret", "") + v.SetDefault("scanner.rescan_interval", "24h") // Log shipper defaults v.SetDefault("log_shipper.batch_size", 100) diff --git a/pkg/hold/pds/scan_broadcaster.go b/pkg/hold/pds/scan_broadcaster.go index d5f205a..ba6c6e4 100644 --- a/pkg/hold/pds/scan_broadcaster.go +++ b/pkg/hold/pds/scan_broadcaster.go @@ -7,7 +7,10 @@ import ( "encoding/hex" "encoding/json" "fmt" + "io" "log/slog" + "net/http" + "net/url" "strings" "sync" "time" @@ -33,6 +36,13 @@ type ScanBroadcaster struct { ackTimeout time.Duration secret string // Shared secret for scanner authentication ownsDB bool // true when this broadcaster opened the connection itself + + // Proactive scan scheduling + rescanInterval time.Duration // Minimum interval between re-scans (0 = disabled) + stopCh chan struct{} // Signal to stop background goroutines + wg sync.WaitGroup // Wait for background goroutines to finish + userIdx int // Round-robin index through users for proactive scanning + predecessorCache map[string]bool // holdDID → "is this hold's successor us?" } // ScanSubscriber represents a connected scanner WebSocket client @@ -80,7 +90,7 @@ type VulnerabilitySummary struct { // NewScanBroadcaster creates a new scan job broadcaster // dbPath should point to a SQLite database file (e.g., "/path/to/pds/db.sqlite3") -func NewScanBroadcaster(holdDID, holdEndpoint, secret, dbPath string, s3svc *s3.S3Service, holdPDS *HoldPDS) (*ScanBroadcaster, error) { +func NewScanBroadcaster(holdDID, holdEndpoint, secret, dbPath string, s3svc *s3.S3Service, holdPDS *HoldPDS, rescanInterval time.Duration) (*ScanBroadcaster, error) { dsn := dbPath if dbPath != ":memory:" && !strings.HasPrefix(dbPath, "file:") { dsn = "file:" + dbPath @@ -107,15 +117,18 @@ func NewScanBroadcaster(holdDID, holdEndpoint, secret, dbPath string, s3svc *s3. } sb := &ScanBroadcaster{ - subscribers: make([]*ScanSubscriber, 0), - db: db, - holdDID: holdDID, - holdEndpoint: holdEndpoint, - s3: s3svc, - pds: holdPDS, - ackTimeout: 5 * time.Minute, - secret: secret, - ownsDB: true, + subscribers: make([]*ScanSubscriber, 0), + db: db, + holdDID: holdDID, + holdEndpoint: holdEndpoint, + s3: s3svc, + pds: holdPDS, + ackTimeout: 5 * time.Minute, + secret: secret, + ownsDB: true, + rescanInterval: rescanInterval, + stopCh: make(chan struct{}), + predecessorCache: make(map[string]bool), } if err := sb.initSchema(); err != nil { @@ -124,32 +137,50 @@ func NewScanBroadcaster(holdDID, holdEndpoint, secret, dbPath string, s3svc *s3. } // Start re-dispatch loop for timed-out jobs + sb.wg.Add(1) go sb.reDispatchLoop() + // Start proactive scan loop if rescan interval is configured + if rescanInterval > 0 { + sb.wg.Add(1) + go sb.proactiveScanLoop() + slog.Info("Proactive scan scheduler started", "rescanInterval", rescanInterval) + } + return sb, nil } // NewScanBroadcasterWithDB creates a scan job broadcaster using an existing *sql.DB connection. // The caller is responsible for the DB lifecycle. -func NewScanBroadcasterWithDB(holdDID, holdEndpoint, secret string, db *sql.DB, s3svc *s3.S3Service, holdPDS *HoldPDS) (*ScanBroadcaster, error) { +func NewScanBroadcasterWithDB(holdDID, holdEndpoint, secret string, db *sql.DB, s3svc *s3.S3Service, holdPDS *HoldPDS, rescanInterval time.Duration) (*ScanBroadcaster, error) { sb := &ScanBroadcaster{ - subscribers: make([]*ScanSubscriber, 0), - db: db, - holdDID: holdDID, - holdEndpoint: holdEndpoint, - s3: s3svc, - pds: holdPDS, - ackTimeout: 5 * time.Minute, - secret: secret, - ownsDB: false, + subscribers: make([]*ScanSubscriber, 0), + db: db, + holdDID: holdDID, + holdEndpoint: holdEndpoint, + s3: s3svc, + pds: holdPDS, + ackTimeout: 5 * time.Minute, + secret: secret, + ownsDB: false, + rescanInterval: rescanInterval, + stopCh: make(chan struct{}), + predecessorCache: make(map[string]bool), } if err := sb.initSchema(); err != nil { return nil, fmt.Errorf("failed to initialize scan_jobs schema: %w", err) } + sb.wg.Add(1) go sb.reDispatchLoop() + if rescanInterval > 0 { + sb.wg.Add(1) + go sb.proactiveScanLoop() + slog.Info("Proactive scan scheduler started", "rescanInterval", rescanInterval) + } + return sb, nil } @@ -587,11 +618,18 @@ func (sb *ScanBroadcaster) drainPendingJobs(sub *ScanSubscriber, cursor int64) { // reDispatchLoop periodically checks for timed-out jobs and re-dispatches them func (sb *ScanBroadcaster) reDispatchLoop() { + defer sb.wg.Done() + ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() - for range ticker.C { - sb.reDispatchTimedOut() + for { + select { + case <-sb.stopCh: + return + case <-ticker.C: + sb.reDispatchTimedOut() + } } } @@ -649,8 +687,12 @@ func (sb *ScanBroadcaster) reDispatchTimedOut() { } } -// Close closes the scan broadcaster's database connection +// Close stops background goroutines and closes the scan broadcaster's database connection func (sb *ScanBroadcaster) Close() error { + if sb.stopCh != nil { + close(sb.stopCh) + sb.wg.Wait() + } if sb.db != nil && sb.ownsDB { return sb.db.Close() } @@ -667,6 +709,288 @@ func (sb *ScanBroadcaster) ValidateScannerSecret(secret string) bool { return sb.secret != "" && secret == sb.secret } +// proactiveScanLoop periodically finds manifests needing scanning and enqueues jobs. +// It fetches manifest records from users' PDS (the source of truth) and creates scan +// jobs for manifests that haven't been scanned recently. +func (sb *ScanBroadcaster) proactiveScanLoop() { + defer sb.wg.Done() + + // Wait a bit before starting to let the system settle + select { + case <-sb.stopCh: + return + case <-time.After(30 * time.Second): + } + + ticker := time.NewTicker(60 * time.Second) + defer ticker.Stop() + + for { + select { + case <-sb.stopCh: + slog.Info("Proactive scan loop stopped") + return + case <-ticker.C: + sb.tryEnqueueProactiveScan() + } + } +} + +// tryEnqueueProactiveScan finds the next manifest needing a scan and enqueues it. +// Only enqueues one job per call to avoid flooding the scanner. +func (sb *ScanBroadcaster) tryEnqueueProactiveScan() { + if !sb.hasConnectedScanners() { + return + } + if sb.hasActiveJobs() { + return + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + + // Get all users who have pushed to this hold + stats, err := sb.pds.ListStats(ctx) + if err != nil { + slog.Error("Proactive scan: failed to list stats", "error", err) + return + } + + // Extract unique user DIDs + seen := make(map[string]bool) + var userDIDs []string + for _, s := range stats { + if !seen[s.OwnerDID] { + seen[s.OwnerDID] = true + userDIDs = append(userDIDs, s.OwnerDID) + } + } + + if len(userDIDs) == 0 { + return + } + + // Round-robin through users, trying each until we find work or exhaust the list + for attempts := 0; attempts < len(userDIDs); attempts++ { + idx := sb.userIdx % len(userDIDs) + sb.userIdx++ + userDID := userDIDs[idx] + + if sb.tryEnqueueForUser(ctx, userDID) { + return // Enqueued one job, done for this tick + } + } +} + +// tryEnqueueForUser fetches manifests from a user's PDS and enqueues a scan for the +// first one that needs scanning. Returns true if a job was enqueued. +func (sb *ScanBroadcaster) tryEnqueueForUser(ctx context.Context, userDID string) bool { + // Resolve user DID to PDS endpoint and handle + did, userHandle, pdsEndpoint, err := atproto.ResolveIdentity(ctx, userDID) + if err != nil { + slog.Debug("Proactive scan: failed to resolve user identity", + "userDID", userDID, "error", err) + return false + } + + // Fetch manifest records from user's PDS + client := atproto.NewClient(pdsEndpoint, did, "") + var cursor string + for { + records, nextCursor, err := client.ListRecordsForRepo(ctx, did, atproto.ManifestCollection, 100, cursor) + if err != nil { + slog.Debug("Proactive scan: failed to list manifest records", + "userDID", did, "pds", pdsEndpoint, "error", err) + return false + } + + for _, record := range records { + var manifest atproto.ManifestRecord + if err := json.Unmarshal(record.Value, &manifest); err != nil { + slog.Debug("Proactive scan: failed to unmarshal manifest record", + "uri", record.URI, "error", err) + continue + } + + // Check if this manifest belongs to us (directly or via successor) + holdDID := manifest.HoldDID + if holdDID == "" { + holdDID = manifest.HoldEndpoint // Legacy field + } + if !sb.isOurManifest(ctx, holdDID) { + continue + } + + // Skip manifest lists (no layers to scan) + if len(manifest.Layers) == 0 { + continue + } + + // Skip if config is nil (shouldn't happen for image manifests, but be safe) + if manifest.Config == nil { + continue + } + + // Check if already scanned recently + if sb.isRecentlyScanned(ctx, manifest.Digest) { + continue + } + + // Construct and enqueue scan job + configJSON, _ := json.Marshal(manifest.Config) + layersJSON, _ := json.Marshal(manifest.Layers) + + slog.Info("Enqueuing proactive scan", + "manifestDigest", manifest.Digest, + "repository", manifest.Repository, + "userDID", did) + + if err := sb.Enqueue(&ScanJobEvent{ + ManifestDigest: manifest.Digest, + Repository: manifest.Repository, + UserDID: did, + UserHandle: userHandle, + Tier: "deckhand", + Config: configJSON, + Layers: layersJSON, + }); err != nil { + slog.Error("Proactive scan: failed to enqueue", + "manifest", manifest.Digest, "error", err) + return false + } + return true + } + + if nextCursor == "" || len(records) == 0 { + break + } + cursor = nextCursor + } + + return false +} + +// isOurManifest checks if a manifest's holdDID matches this hold, either directly +// or via successor (the manifest's hold has set us as its successor). +func (sb *ScanBroadcaster) isOurManifest(ctx context.Context, holdDID string) bool { + if holdDID == "" { + return false + } + + // Direct match + if holdDID == sb.holdDID { + return true + } + + // Check predecessor cache + if isPredecessor, cached := sb.predecessorCache[holdDID]; cached { + return isPredecessor + } + + // Fetch captain record from the other hold's PDS to check successor + isPredecessor := sb.checkPredecessor(ctx, holdDID) + sb.predecessorCache[holdDID] = isPredecessor + return isPredecessor +} + +// checkPredecessor fetches a hold's captain record to check if its successor is us. +func (sb *ScanBroadcaster) checkPredecessor(ctx context.Context, holdDID string) bool { + fetchCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + + holdURL, err := atproto.ResolveHoldURL(fetchCtx, holdDID) + if err != nil { + slog.Debug("Proactive scan: failed to resolve predecessor hold URL", + "holdDID", holdDID, "error", err) + return false + } + + // Fetch captain record: com.atproto.repo.getRecord + recordURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=self", + holdURL, + url.QueryEscape(holdDID), + url.QueryEscape(atproto.CaptainCollection), + ) + + req, err := http.NewRequestWithContext(fetchCtx, "GET", recordURL, nil) + if err != nil { + return false + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + slog.Debug("Proactive scan: failed to fetch predecessor captain record", + "holdDID", holdDID, "error", err) + return false + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return false + } + + body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) // 1MB limit + if err != nil { + return false + } + + var envelope struct { + Value json.RawMessage `json:"value"` + } + if err := json.Unmarshal(body, &envelope); err != nil { + return false + } + + var captain atproto.CaptainRecord + if err := json.Unmarshal(envelope.Value, &captain); err != nil { + return false + } + + if captain.Successor == sb.holdDID { + slog.Info("Proactive scan: discovered predecessor hold", + "predecessorDID", holdDID, "successor", sb.holdDID) + return true + } + + return false +} + +// isRecentlyScanned checks if a manifest has been scanned within the rescan interval. +func (sb *ScanBroadcaster) isRecentlyScanned(ctx context.Context, manifestDigest string) bool { + _, scanRecord, err := sb.pds.GetScanRecord(ctx, manifestDigest) + if err != nil { + return false // Not scanned or error reading → needs scanning + } + + scannedAt, err := time.Parse(time.RFC3339, scanRecord.ScannedAt) + if err != nil { + return false // Can't parse timestamp → treat as needing scan + } + + return time.Since(scannedAt) < sb.rescanInterval +} + +// hasConnectedScanners returns true if at least one scanner is connected. +func (sb *ScanBroadcaster) hasConnectedScanners() bool { + sb.mu.RLock() + defer sb.mu.RUnlock() + return len(sb.subscribers) > 0 +} + +// hasActiveJobs returns true if there are any pending, assigned, or processing scan jobs. +func (sb *ScanBroadcaster) hasActiveJobs() bool { + var count int + err := sb.db.QueryRow(` + SELECT COUNT(*) FROM scan_jobs + WHERE status IN ('pending', 'assigned', 'processing') + `).Scan(&count) + if err != nil { + slog.Error("Failed to check active scan jobs", "error", err) + return true // Assume busy on error + } + return count > 0 +} + func generateSubscriberID() string { b := make([]byte, 8) _, _ = rand.Read(b) diff --git a/pkg/hold/server.go b/pkg/hold/server.go index d13bad4..7d5573d 100644 --- a/pkg/hold/server.go +++ b/pkg/hold/server.go @@ -193,12 +193,13 @@ func NewHoldServer(cfg *Config) (*HoldServer, error) { // Initialize scan broadcaster if scanner secret is configured if cfg.Scanner.Secret != "" { holdDID := s.PDS.DID() + rescanInterval := cfg.Scanner.RescanInterval var sb *pds.ScanBroadcaster if s.holdDB != nil { - sb, err = pds.NewScanBroadcasterWithDB(holdDID, cfg.Server.PublicURL, cfg.Scanner.Secret, s.holdDB.DB, s3Service, s.PDS) + sb, err = pds.NewScanBroadcasterWithDB(holdDID, cfg.Server.PublicURL, cfg.Scanner.Secret, s.holdDB.DB, s3Service, s.PDS, rescanInterval) } else { scanDBPath := cfg.Database.Path + "/db.sqlite3" - sb, err = pds.NewScanBroadcaster(holdDID, cfg.Server.PublicURL, cfg.Scanner.Secret, scanDBPath, s3Service, s.PDS) + sb, err = pds.NewScanBroadcaster(holdDID, cfg.Server.PublicURL, cfg.Scanner.Secret, scanDBPath, s3Service, s.PDS, rescanInterval) } if err != nil { return nil, fmt.Errorf("failed to initialize scan broadcaster: %w", err) @@ -206,7 +207,8 @@ func NewHoldServer(cfg *Config) (*HoldServer, error) { s.scanBroadcaster = sb xrpcHandler.SetScanBroadcaster(sb) ociHandler.SetScanBroadcaster(sb) - slog.Info("Scan broadcaster initialized (scanner WebSocket enabled)") + slog.Info("Scan broadcaster initialized (scanner WebSocket enabled)", + "rescanInterval", rescanInterval) } // Initialize garbage collector