mirrors/at-container-registry
1
0
Fork 0
mirror of https://tangled.org/evan.jarrett.net/at-container-registry synced 2026-08-28 20:06:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix labeler issues

Browse Source
This commit is contained in:
Evan Jarrett
2026-05-09 21:21:20 -05:00
parent 966e391a91
commit 3271ac6dcc
8 changed files with 110 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config-appview.example.yaml
+1 -1
View File
@@ -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:
config-hold.example.yaml
+1 -1
View File
@@ -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
pkg/appview/config.go
+9 -2
View File
@@ -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)
pkg/appview/jetstream/processor.go
+4 -6
View File
@@ -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"`
}
pkg/appview/labeler/subscriber.go
+40 -43
View File
@@ -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)
}
pkg/hold/config.go
+9 -4
View File
@@ -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
pkg/hold/labeler/subscriber.go
+36 -50
View File
@@ -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.
pkg/hold/labeler/subscriber_test.go
+10 -18
View File
@@ -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)
}
})
}