diff --git a/config-appview.example.yaml b/config-appview.example.yaml index ab643ea..33b3c7d 100644 --- a/config-appview.example.yaml +++ b/config-appview.example.yaml @@ -94,7 +94,7 @@ ai: api_key: "" # ATProto labeler for content moderation (DMCA takedowns). labeler: - # DID or URL of the ATProto labeler (e.g., did:web:labeler.atcr.io). Empty disables label filtering. + # DID of the ATProto labeler (did:plc:... or did:web:...). Empty disables label filtering. did: "" # Stripe billing integration (requires -tags billing build). billing: diff --git a/config-hold.example.yaml b/config-hold.example.yaml index 467bc5b..85b5985 100644 --- a/config-hold.example.yaml +++ b/config-hold.example.yaml @@ -139,7 +139,7 @@ scanner: rescan_interval: 168h0m0s # Labeler subscription settings. When configured, the hold consumes takedown labels from the named labeler and purges affected records on receipt; GC consults the cache to gate blob cleanup. Empty subscribe_url disables. labeler: - # DID or URL of the ATProto labeler (e.g., did:web:labeler.atcr.io). Empty disables labeler integration. + # DID of the ATProto labeler (did:plc:... or did:web:...). Empty disables labeler integration. did: "" # Reversibility window for takedowns. Blobs survive this long after a takedown so the action can be reversed. After this window the GC reclaims them. Default: 720h (30 days). grace_window: 720h0m0s diff --git a/pkg/appview/config.go b/pkg/appview/config.go index c95dfb5..ef97dce 100644 --- a/pkg/appview/config.go +++ b/pkg/appview/config.go @@ -11,6 +11,7 @@ import ( "fmt" "net/url" "os" + "strings" "time" "github.com/distribution/distribution/v3/configuration" @@ -147,8 +148,11 @@ type AIConfig struct { // LabelerRefConfig defines the connection to an ATProto labeler service. type LabelerRefConfig struct { - // DID or URL of the labeler service for content moderation. - DID string `yaml:"did" comment:"DID or URL of the ATProto labeler (e.g., did:web:labeler.atcr.io). Empty disables label filtering."` + // DID of the labeler service for content moderation. The HTTP endpoint + // is resolved at runtime via the labeler's #atproto_labeler service entry + // in its DID document (plc.directory for did:plc, /.well-known/did.json + // for did:web). + DID string `yaml:"did" comment:"DID of the ATProto labeler (did:plc:... or did:web:...). Empty disables label filtering."` } // setDefaults registers all default values on the given Viper instance. @@ -280,6 +284,9 @@ func LoadConfig(yamlPath string) (*Config, error) { if cfg.Server.DefaultHoldDID == "" { return nil, fmt.Errorf("server.default_hold_did is required (env: ATCR_SERVER_DEFAULT_HOLD_DID)") } + if cfg.Labeler.DID != "" && !strings.HasPrefix(cfg.Labeler.DID, "did:") { + return nil, fmt.Errorf("labeler.did must be a DID (did:plc:... or did:web:...), got %q", cfg.Labeler.DID) + } // Build distribution config (unchanged) distConfig, err := buildDistributionConfig(cfg, v) diff --git a/pkg/appview/jetstream/processor.go b/pkg/appview/jetstream/processor.go index 6aa2596..6032663 100644 --- a/pkg/appview/jetstream/processor.go +++ b/pkg/appview/jetstream/processor.go @@ -1035,16 +1035,14 @@ func (p *Processor) ProcessAccount(ctx context.Context, did string, active bool, return nil } -// extractRepoFromRecord extracts the repository field from a record's JSON data. -// Returns empty string for collections that don't have a repository field -// (e.g., sailor profile, captain, crew). +// extractRepoFromRecord extracts the repository field from a record's JSON data +// for user-owned collections subject to labeler takedowns. Hold-owned collections +// (io.atcr.hold.*) are intentionally excluded — the labeler doesn't moderate holds. func extractRepoFromRecord(collection string, data []byte) string { switch collection { case atproto.ManifestCollection, atproto.TagCollection, - atproto.RepoPageCollection, - atproto.StatsCollection, - atproto.ScanCollection: + atproto.RepoPageCollection: var rec struct { Repository string `json:"repository"` } diff --git a/pkg/appview/labeler/subscriber.go b/pkg/appview/labeler/subscriber.go index b736a6c..9046c2d 100644 --- a/pkg/appview/labeler/subscriber.go +++ b/pkg/appview/labeler/subscriber.go @@ -4,6 +4,7 @@ package labeler import ( "bytes" + "context" "database/sql" "errors" "fmt" @@ -13,8 +14,10 @@ import ( "time" "atcr.io/pkg/appview/db" + "atcr.io/pkg/atproto" comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/events" "github.com/gorilla/websocket" ) @@ -25,19 +28,19 @@ const TakedownLabelValue = "!takedown" // Subscriber connects to a labeler's subscribeLabels endpoint and mirrors // the current set of active takedowns into the appview database. type Subscriber struct { - labelerURL string labelerDID string database *sql.DB stopCh chan struct{} } -// NewSubscriber creates a new labeler subscriber. labelerDIDOrURL is the -// original config value (used to preserve a configured did:web identifier -// when present); labelerURL is the resolved HTTP(S) endpoint. -func NewSubscriber(labelerDIDOrURL, labelerURL string, database *sql.DB) *Subscriber { +// NewSubscriber creates a new labeler subscriber. labelerDID is a did:plc or +// did:web identifier. The websocket endpoint is resolved on each (re)connect +// via the shared identity directory's #atproto_labeler service entry, so the +// labeler can move (or fix a misconfigured endpoint) without clients +// redeploying. +func NewSubscriber(labelerDID string, database *sql.DB) *Subscriber { return &Subscriber{ - labelerURL: labelerURL, - labelerDID: deriveLabelerDID(labelerDIDOrURL, labelerURL), + labelerDID: labelerDID, database: database, stopCh: make(chan struct{}), } @@ -88,7 +91,14 @@ func (s *Subscriber) connect() error { return fmt.Errorf("failed to get cursor: %w", err) } - wsURL := toWebSocketURL(s.labelerURL) + "/xrpc/com.atproto.label.subscribeLabels" + resolveCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + httpURL, err := resolveLabelerURL(resolveCtx, s.labelerDID) + cancel() + if err != nil { + return fmt.Errorf("resolve labeler endpoint: %w", err) + } + + wsURL := toWebSocketURL(httpURL) + "/xrpc/com.atproto.label.subscribeLabels" if cursor > 0 { wsURL += fmt.Sprintf("?cursor=%d", cursor) } @@ -101,7 +111,7 @@ func (s *Subscriber) connect() error { } defer conn.Close() - slog.Info("Connected to labeler", "url", s.labelerURL) + slog.Info("Connected to labeler", "url", httpURL) for { select { @@ -267,24 +277,25 @@ func decodeFrame(payload []byte) (int64, []*comatproto.LabelDefs_Label, error) { } } -// deriveLabelerDID returns the canonical labeler DID for source filtering. -// When the operator gave us a did:... identifier directly, we use it as-is. -// When they gave us a URL, we derive a did:web from its host so dev URLs -// like http://labeler:5002 yield did:web:labeler%3A5002, matching the -// labeler's own self-served identity. -func deriveLabelerDID(labelerDIDOrURL, httpURL string) string { - if strings.HasPrefix(labelerDIDOrURL, "did:") { - return labelerDIDOrURL - } - u, err := url.Parse(httpURL) +// resolveLabelerURL resolves a labeler DID to its HTTP(S) endpoint by looking +// up the #atproto_labeler service in the shared identity directory: did:plc +// via plc.directory, did:web via /.well-known/did.json. The directory is the +// source of truth — clients don't need redeploying when the labeler moves or +// fixes a misconfigured endpoint. +func resolveLabelerURL(ctx context.Context, labelerDID string) (string, error) { + parsed, err := syntax.ParseDID(labelerDID) if err != nil { - return labelerDIDOrURL + return "", fmt.Errorf("labeler: invalid DID %q: %w", labelerDID, err) } - host := u.Hostname() - if port := u.Port(); port != "" { - host += "%3A" + port + ident, err := atproto.GetDirectory().LookupDID(ctx, parsed) + if err != nil { + return "", fmt.Errorf("labeler: failed to resolve %s: %w", labelerDID, err) } - return "did:web:" + host + endpoint := ident.GetServiceEndpoint("atproto_labeler") + if endpoint == "" { + return "", fmt.Errorf("labeler: %s has no #atproto_labeler service endpoint", labelerDID) + } + return endpoint, nil } // toWebSocketURL converts an HTTP URL to a WebSocket URL. @@ -302,25 +313,11 @@ func toWebSocketURL(httpURL string) string { return u.String() } -// ParseLabelerURL parses a labeler DID or URL into an HTTP URL. -func ParseLabelerURL(labelerDIDOrURL string) string { - if strings.HasPrefix(labelerDIDOrURL, "http://") || strings.HasPrefix(labelerDIDOrURL, "https://") { - return labelerDIDOrURL - } - if strings.HasPrefix(labelerDIDOrURL, "did:web:") { - host := strings.TrimPrefix(labelerDIDOrURL, "did:web:") - host = strings.ReplaceAll(host, "%3A", ":") - return "https://" + host - } - return labelerDIDOrURL -} - -// SubscriberFromConfig creates a Subscriber from a labeler DID/URL config value. -// Returns nil if labelerDIDOrURL is empty. -func SubscriberFromConfig(labelerDIDOrURL string, database *sql.DB) *Subscriber { - if labelerDIDOrURL == "" { +// SubscriberFromConfig creates a Subscriber from a labeler DID config value. +// Returns nil if labelerDID is empty. +func SubscriberFromConfig(labelerDID string, database *sql.DB) *Subscriber { + if labelerDID == "" { return nil } - labelerURL := ParseLabelerURL(labelerDIDOrURL) - return NewSubscriber(labelerDIDOrURL, labelerURL, database) + return NewSubscriber(labelerDID, database) } diff --git a/pkg/hold/config.go b/pkg/hold/config.go index 5f32cd2..cb98374 100644 --- a/pkg/hold/config.go +++ b/pkg/hold/config.go @@ -197,10 +197,10 @@ func (s ServerConfig) AppviewURL() string { // survive a configurable grace window before being collected, preserving // reversibility. type LabelerConfig struct { - // DID or URL of the labeler service. Accepts did:web:... (resolved to - // the corresponding HTTPS host) or a raw http/https URL. Empty disables - // labeler integration. - DID string `yaml:"did" comment:"DID or URL of the ATProto labeler (e.g., did:web:labeler.atcr.io). Empty disables labeler integration."` + // DID of the labeler service (did:plc:... or did:web:...). The HTTP + // endpoint is resolved at runtime via the labeler's #atproto_labeler + // service entry in its DID document. Empty disables labeler integration. + DID string `yaml:"did" comment:"DID of the ATProto labeler (did:plc:... or did:web:...). Empty disables labeler integration."` // Grace window for reversibility. Until a takedown is older than this, // the GC keeps blobs referenced even though their layer records were @@ -386,6 +386,11 @@ func LoadConfig(yamlPath string) (*Config, error) { return nil, fmt.Errorf("database.did_method must be 'web' or 'plc', got %q", cfg.Database.DIDMethod) } + // Validate labeler DID + if cfg.Labeler.DID != "" && !strings.HasPrefix(cfg.Labeler.DID, "did:") { + return nil, fmt.Errorf("labeler.did must be a DID (did:plc:... or did:web:...), got %q", cfg.Labeler.DID) + } + // Store config path for subsystem config loading (e.g. billing) cfg.configPath = yamlPath diff --git a/pkg/hold/labeler/subscriber.go b/pkg/hold/labeler/subscriber.go index 36df006..b5ad8b2 100644 --- a/pkg/hold/labeler/subscriber.go +++ b/pkg/hold/labeler/subscriber.go @@ -10,7 +10,10 @@ import ( "strings" "time" + "atcr.io/pkg/atproto" + comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/events" "github.com/gorilla/websocket" ) @@ -37,28 +40,20 @@ type PurgeOutcome struct { // Subscriber connects to a labeler's subscribeLabels endpoint, mirrors // takedowns into the local cache, and triggers record purges on the hold. type Subscriber struct { - labelerURL string labelerDID string cache *Cache purger Purger stopCh chan struct{} } -// NewSubscriber builds a subscriber for the given labeler. labelerDIDOrURL -// may be either: -// -// - a did:web identifier (e.g. did:web:labeler.atcr.io) → resolved to https://labeler.atcr.io -// - a raw http/https URL (e.g. http://172.28.0.4:5002 for dev) -// -// The websocket URL is derived from the resolved HTTPS endpoint; the -// labeler's DID (used to filter the Src field on incoming labels) is derived -// the same way the appview's labeler subscriber derives it, so a single -// config field suffices. -func NewSubscriber(labelerDIDOrURL string, cache *Cache, purger Purger) *Subscriber { - httpURL := parseLabelerURL(labelerDIDOrURL) +// NewSubscriber builds a subscriber for the given labeler DID (did:plc or +// did:web). The websocket endpoint is resolved on each (re)connect through +// the shared identity directory's #atproto_labeler service entry, so the +// labeler can move (or fix a misconfigured endpoint) without clients +// redeploying. +func NewSubscriber(labelerDID string, cache *Cache, purger Purger) *Subscriber { return &Subscriber{ - labelerURL: httpURL, - labelerDID: deriveLabelerDID(labelerDIDOrURL, httpURL), + labelerDID: labelerDID, cache: cache, purger: purger, stopCh: make(chan struct{}), @@ -90,7 +85,7 @@ func (s *Subscriber) run() { if err := s.connect(); err != nil { slog.Warn("Hold labeler subscription error, reconnecting", - "labeler", s.labelerURL, + "labeler", s.labelerDID, "error", err, "backoff", backoff, ) @@ -114,7 +109,14 @@ func (s *Subscriber) connect() error { return fmt.Errorf("get cursor: %w", err) } - wsURL := toWebSocketURL(s.labelerURL) + "/xrpc/com.atproto.label.subscribeLabels" + resolveCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + httpURL, err := resolveLabelerURL(resolveCtx, s.labelerDID) + cancel() + if err != nil { + return fmt.Errorf("resolve labeler endpoint: %w", err) + } + + wsURL := toWebSocketURL(httpURL) + "/xrpc/com.atproto.label.subscribeLabels" if cursor > 0 { wsURL += fmt.Sprintf("?cursor=%d", cursor) } @@ -125,7 +127,7 @@ func (s *Subscriber) connect() error { return fmt.Errorf("websocket dial: %w", err) } defer conn.Close() - slog.Info("Hold connected to labeler", "url", s.labelerURL) + slog.Info("Hold connected to labeler", "url", httpURL) for { select { @@ -321,41 +323,25 @@ func (s *Subscriber) trustsSource(src string) bool { return src == s.labelerDID } -// parseLabelerURL accepts either a did:web:... identifier or a raw http/https -// URL and returns the HTTPS (or HTTP for did:web pointing at a hostname with -// %3A-encoded port in test mode) endpoint to talk to. did:web hosts with -// %3A-encoded ports are decoded back to colons. Mirrors the appview's -// ParseLabelerURL so a single config field works in both places. -func parseLabelerURL(labelerDIDOrURL string) string { - if strings.HasPrefix(labelerDIDOrURL, "http://") || strings.HasPrefix(labelerDIDOrURL, "https://") { - return labelerDIDOrURL - } - if strings.HasPrefix(labelerDIDOrURL, "did:web:") { - host := strings.TrimPrefix(labelerDIDOrURL, "did:web:") - host = strings.ReplaceAll(host, "%3A", ":") - return "https://" + host - } - return labelerDIDOrURL -} - -// deriveLabelerDID returns the canonical labeler DID for source filtering. -// When the operator gave us a did:web identifier directly, we use it as-is. -// When they gave us a URL, we derive a did:web from its host (so dev URLs -// like http://172.28.0.4:5002 yield did:web:172.28.0.4%3A5002, matching the -// labeler's own self-served identity). -func deriveLabelerDID(labelerDIDOrURL, httpURL string) string { - if strings.HasPrefix(labelerDIDOrURL, "did:") { - return labelerDIDOrURL - } - u, err := url.Parse(httpURL) +// resolveLabelerURL resolves a labeler DID to its HTTP(S) endpoint by looking +// up the #atproto_labeler service in the shared identity directory: did:plc +// via plc.directory, did:web via /.well-known/did.json. The directory is the +// source of truth — clients don't need redeploying when the labeler moves or +// fixes a misconfigured endpoint. +func resolveLabelerURL(ctx context.Context, labelerDID string) (string, error) { + parsed, err := syntax.ParseDID(labelerDID) if err != nil { - return labelerDIDOrURL + return "", fmt.Errorf("labeler: invalid DID %q: %w", labelerDID, err) } - host := u.Hostname() - if port := u.Port(); port != "" { - host += "%3A" + port + ident, err := atproto.GetDirectory().LookupDID(ctx, parsed) + if err != nil { + return "", fmt.Errorf("labeler: failed to resolve %s: %w", labelerDID, err) } - return "did:web:" + host + endpoint := ident.GetServiceEndpoint("atproto_labeler") + if endpoint == "" { + return "", fmt.Errorf("labeler: %s has no #atproto_labeler service endpoint", labelerDID) + } + return endpoint, nil } // toWebSocketURL converts an HTTP URL to a WebSocket URL. http→ws, https→wss. diff --git a/pkg/hold/labeler/subscriber_test.go b/pkg/hold/labeler/subscriber_test.go index fd8216d..2baf55f 100644 --- a/pkg/hold/labeler/subscriber_test.go +++ b/pkg/hold/labeler/subscriber_test.go @@ -142,25 +142,17 @@ func TestApplyLabelIgnoresUntrustedSource(t *testing.T) { } } -func TestSubscriberDerivesDIDFromURL(t *testing.T) { - tests := []struct { - input string - wantURL string - wantDID string - }{ - {"did:web:labeler.atcr.io", "https://labeler.atcr.io", "did:web:labeler.atcr.io"}, - {"did:web:172.28.0.4%3A5002", "https://172.28.0.4:5002", "did:web:172.28.0.4%3A5002"}, - {"http://172.28.0.4:5002", "http://172.28.0.4:5002", "did:web:172.28.0.4%3A5002"}, - {"https://labeler.atcr.io", "https://labeler.atcr.io", "did:web:labeler.atcr.io"}, +func TestSubscriberStoresDID(t *testing.T) { + tests := []string{ + "did:web:labeler.atcr.io", + "did:web:172.28.0.4%3A5002", + "did:plc:4zul2zfigjltl24ti24xj3hy", } - for _, tt := range tests { - t.Run(tt.input, func(t *testing.T) { - sub := NewSubscriber(tt.input, nil, nil) - if sub.labelerURL != tt.wantURL { - t.Errorf("labelerURL = %q, want %q", sub.labelerURL, tt.wantURL) - } - if sub.labelerDID != tt.wantDID { - t.Errorf("labelerDID = %q, want %q", sub.labelerDID, tt.wantDID) + for _, did := range tests { + t.Run(did, func(t *testing.T) { + sub := NewSubscriber(did, nil, nil) + if sub.labelerDID != did { + t.Errorf("labelerDID = %q, want %q", sub.labelerDID, did) } }) }