A container registry that uses the AT Protocol for manifest storage and S3 for blob storage.
first pass at implementing a label service
+2985
-15
+82
cmd/labeler/main.go
+82
cmd/labeler/main.go
···
1
+
package main
2
+
3
+
import (
4
+
"fmt"
5
+
"os"
6
+
7
+
"github.com/spf13/cobra"
8
+
9
+
"atcr.io/pkg/labeler"
10
+
)
11
+
12
+
var configFile string
13
+
14
+
var rootCmd = &cobra.Command{
15
+
Use: "atcr-labeler",
16
+
Short: "ATCR Labeler Service - ATProto content moderation",
17
+
}
18
+
19
+
var serveCmd = &cobra.Command{
20
+
Use: "serve",
21
+
Short: "Start the labeler service",
22
+
Long: `Start the ATCR labeler service with admin UI and subscribeLabels endpoint.
23
+
24
+
Configuration is loaded from the appview config YAML (labeler section).
25
+
Use --config to specify the config file path.`,
26
+
Args: cobra.NoArgs,
27
+
RunE: func(cmd *cobra.Command, args []string) error {
28
+
cfg, err := labeler.LoadConfig(configFile)
29
+
if err != nil {
30
+
return fmt.Errorf("failed to load config: %w", err)
31
+
}
32
+
33
+
server, err := labeler.NewServer(cfg)
34
+
if err != nil {
35
+
return fmt.Errorf("failed to initialize labeler: %w", err)
36
+
}
37
+
38
+
return server.Serve()
39
+
},
40
+
}
41
+
42
+
var configCmd = &cobra.Command{
43
+
Use: "config",
44
+
Short: "Configuration management commands",
45
+
}
46
+
47
+
var configInitCmd = &cobra.Command{
48
+
Use: "init [path]",
49
+
Short: "Generate an example configuration file",
50
+
Long: `Generate an example YAML configuration file with all available options.`,
51
+
Args: cobra.MaximumNArgs(1),
52
+
RunE: func(cmd *cobra.Command, args []string) error {
53
+
yamlBytes, err := labeler.ExampleYAML()
54
+
if err != nil {
55
+
return fmt.Errorf("failed to generate example config: %w", err)
56
+
}
57
+
if len(args) == 1 {
58
+
if err := os.WriteFile(args[0], yamlBytes, 0644); err != nil {
59
+
return fmt.Errorf("failed to write config file: %w", err)
60
+
}
61
+
fmt.Fprintf(os.Stderr, "Wrote example config to %s\n", args[0])
62
+
return nil
63
+
}
64
+
fmt.Print(string(yamlBytes))
65
+
return nil
66
+
},
67
+
}
68
+
69
+
func init() {
70
+
serveCmd.Flags().StringVarP(&configFile, "config", "c", "", "path to YAML configuration file")
71
+
72
+
configCmd.AddCommand(configInitCmd)
73
+
74
+
rootCmd.AddCommand(serveCmd)
75
+
rootCmd.AddCommand(configCmd)
76
+
}
77
+
78
+
func main() {
79
+
if err := rootCmd.Execute(); err != nil {
80
+
os.Exit(1)
81
+
}
82
+
}
+96
-3
deploy/upcloud/cloudinit.go
+96
-3
deploy/upcloud/cloudinit.go
···
28
28
//go:embed configs/scanner.yaml.tmpl
29
29
var scannerConfigTmpl string
30
30
31
+
//go:embed systemd/labeler.service.tmpl
32
+
var labelerServiceTmpl string
33
+
34
+
//go:embed configs/labeler.yaml.tmpl
35
+
var labelerConfigTmpl string
36
+
31
37
//go:embed configs/cloudinit.sh.tmpl
32
38
var cloudInitTmpl string
33
39
···
111
117
return buf.String(), nil
112
118
}
113
119
120
+
// labelerServiceUnitParams holds values for rendering the labeler systemd unit.
121
+
type labelerServiceUnitParams struct {
122
+
DisplayName string // e.g. "Seamark"
123
+
User string // e.g. "seamark"
124
+
BinaryPath string // e.g. "/opt/seamark/bin/seamark-labeler"
125
+
ConfigPath string // e.g. "/etc/seamark/labeler.yaml"
126
+
DataDir string // e.g. "/var/lib/seamark"
127
+
ServiceName string // e.g. "seamark-labeler"
128
+
AppviewServiceName string // e.g. "seamark-appview" (After= dependency)
129
+
}
130
+
131
+
func renderLabelerServiceUnit(p labelerServiceUnitParams) (string, error) {
132
+
t, err := template.New("labeler-service").Parse(labelerServiceTmpl)
133
+
if err != nil {
134
+
return "", fmt.Errorf("parse labeler service template: %w", err)
135
+
}
136
+
var buf bytes.Buffer
137
+
if err := t.Execute(&buf, p); err != nil {
138
+
return "", fmt.Errorf("render labeler service template: %w", err)
139
+
}
140
+
return buf.String(), nil
141
+
}
142
+
114
143
// generateAppviewCloudInit generates the cloud-init user-data script for the appview server.
115
-
// Sets up the OS, directories, config, and systemd unit. Binaries are deployed separately via SCP.
116
-
func generateAppviewCloudInit(cfg *InfraConfig, vals *ConfigValues) (string, error) {
144
+
// When withLabeler is true, a second phase is appended that creates labeler data
145
+
// directories and installs a labeler systemd service. Binaries are deployed separately via SCP.
146
+
func generateAppviewCloudInit(cfg *InfraConfig, vals *ConfigValues, withLabeler bool) (string, error) {
117
147
naming := cfg.Naming()
118
148
119
149
configYAML, err := renderConfig(appviewConfigTmpl, vals)
···
133
163
return "", fmt.Errorf("appview service unit: %w", err)
134
164
}
135
165
136
-
return generateCloudInit(cloudInitParams{
166
+
script, err := generateCloudInit(cloudInitParams{
137
167
BinaryName: naming.Appview(),
138
168
ServiceUnit: serviceUnit,
139
169
ConfigYAML: configYAML,
···
146
176
LogFile: naming.LogFile(),
147
177
DisplayName: naming.DisplayName(),
148
178
})
179
+
if err != nil {
180
+
return "", err
181
+
}
182
+
183
+
if !withLabeler {
184
+
return script, nil
185
+
}
186
+
187
+
// Render labeler config YAML
188
+
labelerConfigYAML, err := renderConfig(labelerConfigTmpl, vals)
189
+
if err != nil {
190
+
return "", fmt.Errorf("labeler config: %w", err)
191
+
}
192
+
193
+
// Append labeler setup phase
194
+
labelerUnit, err := renderLabelerServiceUnit(labelerServiceUnitParams{
195
+
DisplayName: naming.DisplayName(),
196
+
User: naming.SystemUser(),
197
+
BinaryPath: naming.InstallDir() + "/bin/" + naming.Labeler(),
198
+
ConfigPath: naming.LabelerConfigPath(),
199
+
DataDir: naming.BasePath(),
200
+
ServiceName: naming.Labeler(),
201
+
AppviewServiceName: naming.Appview(),
202
+
})
203
+
if err != nil {
204
+
return "", fmt.Errorf("labeler service unit: %w", err)
205
+
}
206
+
207
+
// Escape single quotes for heredoc embedding
208
+
labelerUnit = strings.ReplaceAll(labelerUnit, "'", "'\\''")
209
+
labelerConfigYAML = strings.ReplaceAll(labelerConfigYAML, "'", "'\\''")
210
+
211
+
labelerPhase := fmt.Sprintf(`
212
+
# === Labeler Setup ===
213
+
214
+
# Labeler data dirs
215
+
mkdir -p %s
216
+
chown -R %s:%s %s
217
+
218
+
# Labeler config
219
+
cat > %s << 'CFGEOF'
220
+
%s
221
+
CFGEOF
222
+
223
+
# Labeler systemd service
224
+
cat > /etc/systemd/system/%s.service << 'SVCEOF'
225
+
%s
226
+
SVCEOF
227
+
systemctl daemon-reload
228
+
systemctl enable %s
229
+
230
+
echo "=== Labeler setup complete ==="
231
+
`,
232
+
naming.LabelerDataDir(),
233
+
naming.SystemUser(), naming.SystemUser(), naming.LabelerDataDir(),
234
+
naming.LabelerConfigPath(),
235
+
labelerConfigYAML,
236
+
naming.Labeler(),
237
+
labelerUnit,
238
+
naming.Labeler(),
239
+
)
240
+
241
+
return script + labelerPhase, nil
149
242
}
150
243
151
244
// generateHoldCloudInit generates the cloud-init user-data script for the hold server.
+2
deploy/upcloud/configs/appview.yaml.tmpl
+2
deploy/upcloud/configs/appview.yaml.tmpl
+19
deploy/upcloud/configs/labeler.yaml.tmpl
+19
deploy/upcloud/configs/labeler.yaml.tmpl
···
1
+
version: "0.1"
2
+
log_level: info
3
+
log_shipper:
4
+
backend: ""
5
+
url: ""
6
+
batch_size: 100
7
+
flush_interval: 5s
8
+
username: ""
9
+
password: ""
10
+
labeler:
11
+
enabled: true
12
+
addr: :5002
13
+
owner_did: ""
14
+
db_path: "{{.BasePath}}/labeler/labeler.db"
15
+
server:
16
+
base_url: "https://seamark.dev"
17
+
client_name: Seamark
18
+
client_short_name: Seamark
19
+
test_mode: false
deploy/upcloud/deploy
deploy/upcloud/deploy
This is a binary file and will not be displayed.
+9
deploy/upcloud/naming.go
+9
deploy/upcloud/naming.go
···
57
57
// ScannerDataDir returns the scanner data directory (e.g. "/var/lib/seamark/scanner").
58
58
func (n Naming) ScannerDataDir() string { return n.BasePath() + "/scanner" }
59
59
60
+
// Labeler returns the labeler binary/service name (e.g. "seamark-labeler").
61
+
func (n Naming) Labeler() string { return n.ClientName + "-labeler" }
62
+
63
+
// LabelerConfigPath returns the labeler config file path.
64
+
func (n Naming) LabelerConfigPath() string { return n.ConfigDir() + "/labeler.yaml" }
65
+
66
+
// LabelerDataDir returns the labeler data directory (e.g. "/var/lib/seamark/labeler").
67
+
func (n Naming) LabelerDataDir() string { return n.BasePath() + "/labeler" }
68
+
60
69
// S3Name returns the name used for S3 storage, user, and bucket.
61
70
func (n Naming) S3Name() string { return n.ClientName }
+40
-7
deploy/upcloud/provision.go
+40
-7
deploy/upcloud/provision.go
···
29
29
sshKey, _ := cmd.Flags().GetString("ssh-key")
30
30
s3Secret, _ := cmd.Flags().GetString("s3-secret")
31
31
withScanner, _ := cmd.Flags().GetBool("with-scanner")
32
-
return cmdProvision(token, zone, plan, sshKey, s3Secret, withScanner)
32
+
withLabeler, _ := cmd.Flags().GetBool("with-labeler")
33
+
return cmdProvision(token, zone, plan, sshKey, s3Secret, withScanner, withLabeler)
33
34
},
34
35
}
35
36
···
39
40
provisionCmd.Flags().String("ssh-key", "", "Path to SSH public key file (required)")
40
41
provisionCmd.Flags().String("s3-secret", "", "S3 secret access key (for existing object storage)")
41
42
provisionCmd.Flags().Bool("with-scanner", false, "Deploy vulnerability scanner alongside hold")
43
+
provisionCmd.Flags().Bool("with-labeler", false, "Deploy content moderation labeler alongside appview")
42
44
_ = provisionCmd.MarkFlagRequired("ssh-key")
43
45
rootCmd.AddCommand(provisionCmd)
44
46
}
45
47
46
-
func cmdProvision(token, zone, plan, sshKeyPath, s3Secret string, withScanner bool) error {
48
+
func cmdProvision(token, zone, plan, sshKeyPath, s3Secret string, withScanner, withLabeler bool) error {
47
49
cfg, err := loadConfig(zone, plan, sshKeyPath, s3Secret)
48
50
if err != nil {
49
51
return err
···
95
97
state.ScannerSecret = secret
96
98
fmt.Printf("Generated scanner shared secret\n")
97
99
}
100
+
_ = saveState(state)
101
+
}
102
+
103
+
// Labeler setup
104
+
if withLabeler {
105
+
state.LabelerEnabled = true
98
106
_ = saveState(state)
99
107
}
100
108
···
198
206
appviewCreated := false
199
207
if state.Appview.UUID != "" {
200
208
fmt.Printf("Appview: %s (exists)\n", state.Appview.UUID)
201
-
appviewScript, err := generateAppviewCloudInit(cfg, vals)
209
+
appviewScript, err := generateAppviewCloudInit(cfg, vals, state.LabelerEnabled)
202
210
if err != nil {
203
211
return err
204
212
}
···
212
220
if err := syncConfigKeys("appview", state.Appview.PublicIP, naming.AppviewConfigPath(), appviewConfigYAML); err != nil {
213
221
return fmt.Errorf("appview config sync: %w", err)
214
222
}
223
+
if state.LabelerEnabled {
224
+
labelerConfigYAML, err := renderConfig(labelerConfigTmpl, vals)
225
+
if err != nil {
226
+
return fmt.Errorf("render labeler config: %w", err)
227
+
}
228
+
if err := syncConfigKeys("labeler", state.Appview.PublicIP, naming.LabelerConfigPath(), labelerConfigYAML); err != nil {
229
+
return fmt.Errorf("labeler config sync: %w", err)
230
+
}
231
+
}
215
232
} else {
216
233
fmt.Println("Creating appview server...")
217
-
appviewUserData, err := generateAppviewCloudInit(cfg, vals)
234
+
appviewUserData, err := generateAppviewCloudInit(cfg, vals, state.LabelerEnabled)
218
235
if err != nil {
219
236
return err
220
237
}
···
338
355
if err := buildLocal(rootDir, outputPath, "./cmd/appview"); err != nil {
339
356
return fmt.Errorf("build appview: %w", err)
340
357
}
358
+
if state.LabelerEnabled {
359
+
outputPath := filepath.Join(rootDir, "bin", "atcr-labeler")
360
+
if err := buildLocal(rootDir, outputPath, "./cmd/labeler"); err != nil {
361
+
return fmt.Errorf("build labeler: %w", err)
362
+
}
363
+
}
341
364
}
342
365
if holdCreated {
343
366
outputPath := filepath.Join(rootDir, "bin", "atcr-hold")
···
371
394
if err := scpFile(localPath, state.Appview.PublicIP, remotePath); err != nil {
372
395
return fmt.Errorf("upload appview: %w", err)
373
396
}
397
+
if state.LabelerEnabled {
398
+
labelerLocal := filepath.Join(rootDir, "bin", "atcr-labeler")
399
+
labelerRemote := naming.InstallDir() + "/bin/" + naming.Labeler()
400
+
if err := scpFile(labelerLocal, state.Appview.PublicIP, labelerRemote); err != nil {
401
+
return fmt.Errorf("upload labeler: %w", err)
402
+
}
403
+
}
374
404
}
375
405
if holdCreated {
376
406
localPath := filepath.Join(rootDir, "bin", "atcr-hold")
···
411
441
} else {
412
442
fmt.Println(" 1. Start services:")
413
443
}
444
+
services := []string{naming.Appview(), naming.Hold()}
414
445
if state.ScannerEnabled {
415
-
fmt.Printf(" systemctl start %s / %s / %s\n", naming.Appview(), naming.Hold(), naming.Scanner())
416
-
} else {
417
-
fmt.Printf(" systemctl start %s / %s\n", naming.Appview(), naming.Hold())
446
+
services = append(services, naming.Scanner())
447
+
}
448
+
if state.LabelerEnabled {
449
+
services = append(services, naming.Labeler())
418
450
}
451
+
fmt.Printf(" systemctl start %s\n", strings.Join(services, " / "))
419
452
fmt.Println(" 2. Configure DNS records above")
420
453
421
454
return nil
+1
deploy/upcloud/state.go
+1
deploy/upcloud/state.go
···
20
20
ObjectStorage ObjectStorageState `json:"object_storage"`
21
21
ScannerEnabled bool `json:"scanner_enabled,omitempty"`
22
22
ScannerSecret string `json:"scanner_secret,omitempty"`
23
+
LabelerEnabled bool `json:"labeler_enabled,omitempty"`
23
24
}
24
25
25
26
// Naming returns a Naming helper, defaulting to "seamark" if ClientName is empty.
+25
deploy/upcloud/systemd/labeler.service.tmpl
+25
deploy/upcloud/systemd/labeler.service.tmpl
···
1
+
[Unit]
2
+
Description={{.DisplayName}} Labeler (Content Moderation)
3
+
After=network-online.target {{.AppviewServiceName}}.service
4
+
Wants=network-online.target
5
+
6
+
[Service]
7
+
Type=simple
8
+
User={{.User}}
9
+
Group={{.User}}
10
+
ExecStart={{.BinaryPath}} serve --config {{.ConfigPath}}
11
+
Restart=on-failure
12
+
RestartSec=10
13
+
14
+
ReadWritePaths={{.DataDir}}
15
+
ProtectSystem=strict
16
+
ProtectHome=yes
17
+
NoNewPrivileges=yes
18
+
PrivateTmp=yes
19
+
20
+
StandardOutput=journal
21
+
StandardError=journal
22
+
SyslogIdentifier={{.ServiceName}}
23
+
24
+
[Install]
25
+
WantedBy=multi-user.target
+79
-4
deploy/upcloud/update.go
+79
-4
deploy/upcloud/update.go
···
24
24
target = args[0]
25
25
}
26
26
withScanner, _ := cmd.Flags().GetBool("with-scanner")
27
-
return cmdUpdate(target, withScanner)
27
+
withLabeler, _ := cmd.Flags().GetBool("with-labeler")
28
+
return cmdUpdate(target, withScanner, withLabeler)
28
29
},
29
30
}
30
31
···
40
41
41
42
func init() {
42
43
updateCmd.Flags().Bool("with-scanner", false, "Enable and deploy vulnerability scanner alongside hold")
44
+
updateCmd.Flags().Bool("with-labeler", false, "Enable and deploy content moderation labeler alongside appview")
43
45
rootCmd.AddCommand(updateCmd)
44
46
rootCmd.AddCommand(sshCmd)
45
47
}
46
48
47
-
func cmdUpdate(target string, withScanner bool) error {
49
+
func cmdUpdate(target string, withScanner, withLabeler bool) error {
48
50
state, err := loadState()
49
51
if err != nil {
50
52
return err
···
64
66
state.ScannerSecret = secret
65
67
fmt.Printf("Generated scanner shared secret\n")
66
68
}
69
+
_ = saveState(state)
70
+
}
71
+
72
+
// Enable labeler retroactively via --with-labeler on update
73
+
if withLabeler && !state.LabelerEnabled {
74
+
state.LabelerEnabled = true
67
75
_ = saveState(state)
68
76
}
69
77
···
144
152
}
145
153
}
146
154
155
+
// Build labeler locally if needed
156
+
needLabeler := false
157
+
for _, name := range toUpdate {
158
+
if name == "appview" && state.LabelerEnabled {
159
+
needLabeler = true
160
+
break
161
+
}
162
+
}
163
+
if needLabeler {
164
+
outputPath := filepath.Join(rootDir, "bin", "atcr-labeler")
165
+
if err := buildLocal(rootDir, outputPath, "./cmd/labeler"); err != nil {
166
+
return fmt.Errorf("build labeler: %w", err)
167
+
}
168
+
}
169
+
147
170
// Deploy each target
148
171
for _, name := range toUpdate {
149
172
t := targets[name]
···
244
267
`
245
268
}
246
269
270
+
// Labeler additions for appview server
271
+
labelerRestart := ""
272
+
if name == "appview" && state.LabelerEnabled {
273
+
// Sync labeler config keys
274
+
labelerConfigYAML, err := renderConfig(labelerConfigTmpl, vals)
275
+
if err != nil {
276
+
return fmt.Errorf("render labeler config: %w", err)
277
+
}
278
+
if err := syncConfigKeys("labeler", t.ip, naming.LabelerConfigPath(), labelerConfigYAML); err != nil {
279
+
return fmt.Errorf("labeler config sync: %w", err)
280
+
}
281
+
282
+
// Sync labeler service unit
283
+
labelerUnit, err := renderLabelerServiceUnit(labelerServiceUnitParams{
284
+
DisplayName: naming.DisplayName(),
285
+
User: naming.SystemUser(),
286
+
BinaryPath: naming.InstallDir() + "/bin/" + naming.Labeler(),
287
+
ConfigPath: naming.LabelerConfigPath(),
288
+
DataDir: naming.BasePath(),
289
+
ServiceName: naming.Labeler(),
290
+
AppviewServiceName: naming.Appview(),
291
+
})
292
+
if err != nil {
293
+
return fmt.Errorf("render labeler service unit: %w", err)
294
+
}
295
+
labelerUnitChanged, err := syncServiceUnit("labeler", t.ip, naming.Labeler(), labelerUnit)
296
+
if err != nil {
297
+
return fmt.Errorf("labeler service unit sync: %w", err)
298
+
}
299
+
if labelerUnitChanged {
300
+
daemonReload = "systemctl daemon-reload"
301
+
}
302
+
303
+
// Upload labeler binary
304
+
labelerLocal := filepath.Join(rootDir, "bin", "atcr-labeler")
305
+
labelerRemote := naming.InstallDir() + "/bin/" + naming.Labeler()
306
+
if err := scpFile(labelerLocal, t.ip, labelerRemote); err != nil {
307
+
return fmt.Errorf("upload labeler: %w", err)
308
+
}
309
+
310
+
// Ensure labeler data dirs exist
311
+
labelerSetup := fmt.Sprintf(`mkdir -p %s
312
+
chown -R %s:%s %s`,
313
+
naming.LabelerDataDir(),
314
+
naming.SystemUser(), naming.SystemUser(), naming.LabelerDataDir())
315
+
if _, err := runSSH(t.ip, labelerSetup, false); err != nil {
316
+
return fmt.Errorf("labeler dir setup: %w", err)
317
+
}
318
+
319
+
labelerRestart = fmt.Sprintf("\nsystemctl restart %s", naming.Labeler())
320
+
}
321
+
247
322
// Restart services and health check
248
323
restartScript := fmt.Sprintf(`set -euo pipefail
249
324
%s
250
-
systemctl restart %s%s
325
+
systemctl restart %s%s%s
251
326
sleep 2
252
327
curl -sf %s > /dev/null && echo "HEALTH_OK" || echo "HEALTH_FAIL"
253
-
%s`, daemonReload, t.serviceName, scannerRestart, t.healthURL, scannerHealthCheck)
328
+
%s`, daemonReload, t.serviceName, scannerRestart, labelerRestart, t.healthURL, scannerHealthCheck)
254
329
255
330
output, err := runSSH(t.ip, restartScript, true)
256
331
if err != nil {
+10
pkg/appview/config.go
+10
pkg/appview/config.go
···
32
32
Auth AuthConfig `yaml:"auth" comment:"JWT authentication settings."`
33
33
CredentialHelper CredentialHelperConfig `yaml:"credential_helper" comment:"Credential helper download settings."`
34
34
Legal LegalConfig `yaml:"legal" comment:"Legal page customization for self-hosted instances."`
35
+
Labeler LabelerRefConfig `yaml:"labeler" comment:"ATProto labeler for content moderation (DMCA takedowns)."`
35
36
Billing billing.Config `yaml:"billing" comment:"Stripe billing integration (requires -tags billing build)."`
36
37
Distribution *configuration.Configuration `yaml:"-"` // Wrapped distribution config for compatibility
37
38
}
···
140
141
Jurisdiction string `yaml:"jurisdiction" comment:"Governing law jurisdiction for legal terms."`
141
142
}
142
143
144
+
// LabelerRefConfig defines the connection to an ATProto labeler service.
145
+
type LabelerRefConfig struct {
146
+
// DID or URL of the labeler service for content moderation.
147
+
DID string `yaml:"did" comment:"DID or URL of the ATProto labeler (e.g., did:web:labeler.atcr.io). Empty disables label filtering."`
148
+
}
149
+
143
150
// setDefaults registers all default values on the given Viper instance.
144
151
func setDefaults(v *viper.Viper) {
145
152
v.SetDefault("version", "0.1")
···
192
199
// Legal defaults
193
200
v.SetDefault("legal.company_name", "")
194
201
v.SetDefault("legal.jurisdiction", "")
202
+
203
+
// Labeler defaults
204
+
v.SetDefault("labeler.did", "")
195
205
196
206
// Log formatter (used by distribution config, not in Config struct)
197
207
v.SetDefault("log_formatter", "text")
+79
pkg/appview/db/labels.go
+79
pkg/appview/db/labels.go
···
1
+
package db
2
+
3
+
import (
4
+
"database/sql"
5
+
"time"
6
+
)
7
+
8
+
// LabelChecker wraps a database connection to check takedown labels.
9
+
// Implements middleware.LabelChecker interface.
10
+
type LabelChecker struct {
11
+
db *sql.DB
12
+
}
13
+
14
+
// NewLabelChecker creates a new LabelChecker.
15
+
func NewLabelChecker(database *sql.DB) *LabelChecker {
16
+
return &LabelChecker{db: database}
17
+
}
18
+
19
+
// IsTakenDown checks if a (DID, repository) pair has an active takedown label.
20
+
func (lc *LabelChecker) IsTakenDown(did, repository string) (bool, error) {
21
+
return IsTakenDown(lc.db, did, repository)
22
+
}
23
+
24
+
// Label represents an ATProto label mirrored from a labeler service.
25
+
type Label struct {
26
+
ID int64
27
+
Src string
28
+
URI string
29
+
Val string
30
+
Neg bool
31
+
Cts time.Time
32
+
SubjectDID string
33
+
SubjectRepo string
34
+
Seq int64
35
+
}
36
+
37
+
// IsTakenDown checks if a (DID, repository) pair has an active !takedown label.
38
+
// Also matches user-level labels (subject_repo = ”).
39
+
func IsTakenDown(db DBTX, did, repository string) (bool, error) {
40
+
var exists bool
41
+
err := db.QueryRow(
42
+
`SELECT EXISTS(
43
+
SELECT 1 FROM labels l1
44
+
WHERE l1.subject_did = ?
45
+
AND (l1.subject_repo = ? OR l1.subject_repo = '')
46
+
AND l1.val = '!takedown' AND l1.neg = 0
47
+
AND NOT EXISTS (
48
+
SELECT 1 FROM labels l2
49
+
WHERE l2.src = l1.src AND l2.uri = l1.uri AND l2.val = l1.val
50
+
AND l2.neg = 1 AND l2.id > l1.id
51
+
)
52
+
AND (l1.exp IS NULL OR l1.exp > CURRENT_TIMESTAMP)
53
+
)`,
54
+
did, repository,
55
+
).Scan(&exists)
56
+
return exists, err
57
+
}
58
+
59
+
// UpsertLabel inserts or updates a label from a labeler subscription.
60
+
func UpsertLabel(db DBTX, l *Label) error {
61
+
_, err := db.Exec(
62
+
`INSERT INTO labels (src, uri, val, neg, cts, subject_did, subject_repo, seq)
63
+
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
64
+
ON CONFLICT(src, uri, val, neg) DO UPDATE SET cts = excluded.cts, seq = excluded.seq`,
65
+
l.Src, l.URI, l.Val, l.Neg, l.Cts.UTC().Format(time.RFC3339),
66
+
l.SubjectDID, l.SubjectRepo, l.Seq,
67
+
)
68
+
return err
69
+
}
70
+
71
+
// GetLabelCursor returns the latest sequence number for a given labeler source.
72
+
func GetLabelCursor(db DBTX, src string) (int64, error) {
73
+
var cursor int64
74
+
err := db.QueryRow(
75
+
`SELECT COALESCE(MAX(seq), 0) FROM labels WHERE src = ?`,
76
+
src,
77
+
).Scan(&cursor)
78
+
return cursor, err
79
+
}
+16
pkg/appview/db/migrations/0017_create_labels.yaml
+16
pkg/appview/db/migrations/0017_create_labels.yaml
···
1
+
description: Create labels table for ATProto content moderation (takedowns)
2
+
query: |
3
+
CREATE TABLE IF NOT EXISTS labels (
4
+
id INTEGER PRIMARY KEY AUTOINCREMENT,
5
+
src TEXT NOT NULL,
6
+
uri TEXT NOT NULL,
7
+
val TEXT NOT NULL,
8
+
neg BOOLEAN NOT NULL DEFAULT 0,
9
+
cts TIMESTAMP NOT NULL,
10
+
subject_did TEXT NOT NULL,
11
+
subject_repo TEXT NOT NULL DEFAULT '',
12
+
seq INTEGER NOT NULL DEFAULT 0,
13
+
UNIQUE(src, uri, val, neg)
14
+
);
15
+
CREATE INDEX IF NOT EXISTS idx_labels_subject ON labels(subject_did, subject_repo);
16
+
CREATE INDEX IF NOT EXISTS idx_labels_val ON labels(val);
+16
-1
pkg/appview/db/queries.go
+16
-1
pkg/appview/db/queries.go
···
74
74
SELECT DISTINCT lm.did, lm.repository, lm.latest_id
75
75
FROM latest_manifests lm
76
76
JOIN users u ON lm.did = u.did
77
-
WHERE u.handle LIKE ? ESCAPE '\'
77
+
WHERE (u.handle LIKE ? ESCAPE '\'
78
78
OR u.did = ?
79
79
OR lm.repository LIKE ? ESCAPE '\'
80
80
OR EXISTS (
81
81
SELECT 1 FROM repository_annotations ra
82
82
WHERE ra.did = lm.did AND ra.repository = lm.repository
83
83
AND ra.value LIKE ? ESCAPE '\'
84
+
))
85
+
AND NOT EXISTS (
86
+
SELECT 1 FROM labels
87
+
WHERE (subject_did = lm.did AND (subject_repo = lm.repository OR subject_repo = ''))
88
+
AND val = '!takedown' AND neg = 0
84
89
)
85
90
),
86
91
repo_stats AS (
···
1953
1958
JOIN users u ON m.did = u.did
1954
1959
LEFT JOIN repository_stats rs ON m.did = rs.did AND m.repository = rs.repository
1955
1960
LEFT JOIN repo_pages rp ON m.did = rp.did AND m.repository = rp.repository
1961
+
WHERE NOT EXISTS (
1962
+
SELECT 1 FROM labels
1963
+
WHERE (subject_did = m.did AND (subject_repo = m.repository OR subject_repo = ''))
1964
+
AND val = '!takedown' AND neg = 0
1965
+
)
1956
1966
ORDER BY ` + orderBy + `
1957
1967
LIMIT ?
1958
1968
`
···
2026
2036
JOIN users u ON m.did = u.did
2027
2037
LEFT JOIN repository_stats rs ON m.did = rs.did AND m.repository = rs.repository
2028
2038
LEFT JOIN repo_pages rp ON m.did = rp.did AND m.repository = rp.repository
2039
+
WHERE NOT EXISTS (
2040
+
SELECT 1 FROM labels
2041
+
WHERE (subject_did = m.did AND (subject_repo = m.repository OR subject_repo = ''))
2042
+
AND val = '!takedown' AND neg = 0
2043
+
)
2029
2044
ORDER BY MAX(rs.last_push, m.created_at) DESC
2030
2045
`
2031
2046
+15
pkg/appview/db/schema.sql
+15
pkg/appview/db/schema.sql
···
271
271
PRIMARY KEY(hold_did, manifest_digest)
272
272
);
273
273
CREATE INDEX IF NOT EXISTS idx_scans_user ON scans(user_did);
274
+
275
+
CREATE TABLE IF NOT EXISTS labels (
276
+
id INTEGER PRIMARY KEY AUTOINCREMENT,
277
+
src TEXT NOT NULL,
278
+
uri TEXT NOT NULL,
279
+
val TEXT NOT NULL,
280
+
neg BOOLEAN NOT NULL DEFAULT 0,
281
+
cts TIMESTAMP NOT NULL,
282
+
subject_did TEXT NOT NULL,
283
+
subject_repo TEXT NOT NULL DEFAULT '',
284
+
seq INTEGER NOT NULL DEFAULT 0,
285
+
UNIQUE(src, uri, val, neg)
286
+
);
287
+
CREATE INDEX IF NOT EXISTS idx_labels_subject ON labels(subject_did, subject_repo);
288
+
CREATE INDEX IF NOT EXISTS idx_labels_val ON labels(val);
+6
pkg/appview/handlers/digest.go
+6
pkg/appview/handlers/digest.go
···
45
45
return
46
46
}
47
47
48
+
// Check for takedown labels
49
+
if taken, _ := db.IsTakenDown(h.ReadOnlyDB, did, repository); taken {
50
+
RenderNotFound(w, r, &h.BaseUIHandler)
51
+
return
52
+
}
53
+
48
54
owner, err := db.GetUserByDID(h.ReadOnlyDB, did)
49
55
if err != nil || owner == nil {
50
56
RenderNotFound(w, r, &h.BaseUIHandler)
+6
pkg/appview/handlers/repository.go
+6
pkg/appview/handlers/repository.go
···
34
34
return
35
35
}
36
36
37
+
// Check for takedown labels
38
+
if taken, _ := db.IsTakenDown(h.ReadOnlyDB, did, repository); taken {
39
+
RenderNotFound(w, r, &h.BaseUIHandler)
40
+
return
41
+
}
42
+
37
43
// Look up user by DID
38
44
owner, err := db.GetUserByDID(h.ReadOnlyDB, did)
39
45
if err != nil {
+34
pkg/appview/jetstream/processor.go
+34
pkg/appview/jetstream/processor.go
···
229
229
}
230
230
}
231
231
232
+
// Skip ingestion for taken-down content
233
+
if !isDelete && data != nil {
234
+
if repo := extractRepoFromRecord(collection, data); repo != "" {
235
+
if taken, _ := db.IsTakenDown(p.db, did, repo); taken {
236
+
slog.Debug("Skipping taken-down content",
237
+
"component", "processor",
238
+
"did", did,
239
+
"collection", collection,
240
+
"repository", repo)
241
+
return nil
242
+
}
243
+
}
244
+
}
245
+
232
246
// User-activity collections create/update user entries
233
247
// Skip for deletes - user should already exist, and we don't need to resolve identity
234
248
if !isDelete {
···
971
985
972
986
return nil
973
987
}
988
+
989
+
// extractRepoFromRecord extracts the repository field from a record's JSON data.
990
+
// Returns empty string for collections that don't have a repository field
991
+
// (e.g., sailor profile, captain, crew).
992
+
func extractRepoFromRecord(collection string, data []byte) string {
993
+
switch collection {
994
+
case atproto.ManifestCollection,
995
+
atproto.TagCollection,
996
+
atproto.RepoPageCollection,
997
+
atproto.StatsCollection,
998
+
atproto.ScanCollection:
999
+
var rec struct {
1000
+
Repository string `json:"repository"`
1001
+
}
1002
+
if err := json.Unmarshal(data, &rec); err == nil {
1003
+
return rec.Repository
1004
+
}
1005
+
}
1006
+
return ""
1007
+
}
+239
pkg/appview/labeler/subscriber.go
+239
pkg/appview/labeler/subscriber.go
···
1
+
// Package labeler provides a subscription client for consuming labels
2
+
// from an ATProto labeler service.
3
+
package labeler
4
+
5
+
import (
6
+
"database/sql"
7
+
"encoding/json"
8
+
"fmt"
9
+
"log/slog"
10
+
"net/url"
11
+
"strings"
12
+
"time"
13
+
14
+
"atcr.io/pkg/appview/db"
15
+
16
+
"github.com/gorilla/websocket"
17
+
)
18
+
19
+
// LabelsMessage is the wire format for subscribeLabels events.
20
+
type LabelsMessage struct {
21
+
Seq int64 `json:"seq"`
22
+
Labels []LabelEvent `json:"labels"`
23
+
}
24
+
25
+
// LabelEvent is a single label from the labeler.
26
+
type LabelEvent struct {
27
+
Src string `json:"src"`
28
+
URI string `json:"uri"`
29
+
CID string `json:"cid,omitempty"`
30
+
Val string `json:"val"`
31
+
Neg bool `json:"neg"`
32
+
Cts string `json:"cts"`
33
+
Exp string `json:"exp,omitempty"`
34
+
}
35
+
36
+
// Subscriber connects to a labeler's subscribeLabels endpoint
37
+
// and mirrors labels into the appview database.
38
+
type Subscriber struct {
39
+
labelerURL string
40
+
database *sql.DB
41
+
stopCh chan struct{}
42
+
}
43
+
44
+
// NewSubscriber creates a new labeler subscriber.
45
+
func NewSubscriber(labelerURL string, database *sql.DB) *Subscriber {
46
+
return &Subscriber{
47
+
labelerURL: labelerURL,
48
+
database: database,
49
+
stopCh: make(chan struct{}),
50
+
}
51
+
}
52
+
53
+
// Start begins the subscription loop in a goroutine.
54
+
func (s *Subscriber) Start() {
55
+
go s.run()
56
+
}
57
+
58
+
// Stop signals the subscriber to shut down.
59
+
func (s *Subscriber) Stop() {
60
+
close(s.stopCh)
61
+
}
62
+
63
+
func (s *Subscriber) run() {
64
+
backoff := time.Second
65
+
66
+
for {
67
+
select {
68
+
case <-s.stopCh:
69
+
return
70
+
default:
71
+
}
72
+
73
+
if err := s.connect(); err != nil {
74
+
slog.Warn("Labeler subscription error, reconnecting",
75
+
"error", err,
76
+
"backoff", backoff,
77
+
)
78
+
select {
79
+
case <-s.stopCh:
80
+
return
81
+
case <-time.After(backoff):
82
+
}
83
+
if backoff < 30*time.Second {
84
+
backoff *= 2
85
+
}
86
+
} else {
87
+
backoff = time.Second
88
+
}
89
+
}
90
+
}
91
+
92
+
func (s *Subscriber) connect() error {
93
+
// Get cursor from DB
94
+
// Use the labeler URL as src identifier
95
+
labelerDID := extractDIDFromURL(s.labelerURL)
96
+
cursor, err := db.GetLabelCursor(s.database, labelerDID)
97
+
if err != nil {
98
+
return fmt.Errorf("failed to get cursor: %w", err)
99
+
}
100
+
101
+
// Build WebSocket URL
102
+
wsURL := toWebSocketURL(s.labelerURL) + "/xrpc/com.atproto.label.subscribeLabels"
103
+
if cursor > 0 {
104
+
wsURL += fmt.Sprintf("?cursor=%d", cursor)
105
+
}
106
+
107
+
slog.Info("Connecting to labeler", "url", wsURL, "cursor", cursor)
108
+
109
+
conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
110
+
if err != nil {
111
+
return fmt.Errorf("websocket dial failed: %w", err)
112
+
}
113
+
defer conn.Close()
114
+
115
+
slog.Info("Connected to labeler", "url", s.labelerURL)
116
+
117
+
for {
118
+
select {
119
+
case <-s.stopCh:
120
+
return nil
121
+
default:
122
+
}
123
+
124
+
var msg LabelsMessage
125
+
if err := conn.ReadJSON(&msg); err != nil {
126
+
return fmt.Errorf("read error: %w", err)
127
+
}
128
+
129
+
for _, le := range msg.Labels {
130
+
cts, _ := time.Parse(time.RFC3339, le.Cts)
131
+
did, repo := extractSubjectFromURI(le.URI)
132
+
133
+
label := &db.Label{
134
+
Src: le.Src,
135
+
URI: le.URI,
136
+
Val: le.Val,
137
+
Neg: le.Neg,
138
+
Cts: cts,
139
+
SubjectDID: did,
140
+
SubjectRepo: repo,
141
+
Seq: msg.Seq,
142
+
}
143
+
144
+
if err := db.UpsertLabel(s.database, label); err != nil {
145
+
slog.Warn("Failed to upsert label", "uri", le.URI, "error", err)
146
+
continue
147
+
}
148
+
149
+
slog.Info("Mirrored label",
150
+
"uri", le.URI,
151
+
"val", le.Val,
152
+
"neg", le.Neg,
153
+
"subject_did", did,
154
+
"subject_repo", repo,
155
+
)
156
+
}
157
+
}
158
+
}
159
+
160
+
// extractSubjectFromURI extracts the DID and repository from an AT URI.
161
+
// Examples:
162
+
//
163
+
// at://did:plc:xyz → (did:plc:xyz, "")
164
+
// at://did:plc:xyz/io.atcr.manifest/abc → (did:plc:xyz, "") - repo extracted from record
165
+
// at://did:plc:xyz/io.atcr.repo/myimage → (did:plc:xyz, "myimage")
166
+
func extractSubjectFromURI(uri string) (did, repo string) {
167
+
trimmed := strings.TrimPrefix(uri, "at://")
168
+
parts := strings.SplitN(trimmed, "/", 3)
169
+
if len(parts) == 0 {
170
+
return "", ""
171
+
}
172
+
did = parts[0]
173
+
174
+
// For repo-level summary labels: at://did/io.atcr.repo/reponame
175
+
if len(parts) >= 3 && parts[1] == "io.atcr.repo" {
176
+
repo = parts[2]
177
+
}
178
+
return did, repo
179
+
}
180
+
181
+
// extractDIDFromURL derives a did:web from a labeler URL.
182
+
func extractDIDFromURL(labelerURL string) string {
183
+
u, err := url.Parse(labelerURL)
184
+
if err != nil {
185
+
return labelerURL
186
+
}
187
+
host := u.Hostname()
188
+
if port := u.Port(); port != "" {
189
+
host += "%3A" + port
190
+
}
191
+
return "did:web:" + host
192
+
}
193
+
194
+
// toWebSocketURL converts an HTTP URL to a WebSocket URL.
195
+
func toWebSocketURL(httpURL string) string {
196
+
u, err := url.Parse(httpURL)
197
+
if err != nil {
198
+
return httpURL
199
+
}
200
+
switch u.Scheme {
201
+
case "https":
202
+
u.Scheme = "wss"
203
+
default:
204
+
u.Scheme = "ws"
205
+
}
206
+
return u.String()
207
+
}
208
+
209
+
// ParseLabelerURL parses a labeler DID or URL into an HTTP URL.
210
+
func ParseLabelerURL(labelerDIDOrURL string) string {
211
+
if strings.HasPrefix(labelerDIDOrURL, "http://") || strings.HasPrefix(labelerDIDOrURL, "https://") {
212
+
return labelerDIDOrURL
213
+
}
214
+
if strings.HasPrefix(labelerDIDOrURL, "did:web:") {
215
+
host := strings.TrimPrefix(labelerDIDOrURL, "did:web:")
216
+
host = strings.ReplaceAll(host, "%3A", ":")
217
+
return "https://" + host
218
+
}
219
+
return labelerDIDOrURL
220
+
}
221
+
222
+
// SubscriberFromConfig creates a Subscriber from a labeler DID/URL config value.
223
+
// Returns nil if labelerDIDOrURL is empty.
224
+
func SubscriberFromConfig(labelerDIDOrURL string, database *sql.DB) *Subscriber {
225
+
if labelerDIDOrURL == "" {
226
+
return nil
227
+
}
228
+
labelerURL := ParseLabelerURL(labelerDIDOrURL)
229
+
return NewSubscriber(labelerURL, database)
230
+
}
231
+
232
+
// DecodeLabelsFromJSON decodes a JSON-encoded labels message.
233
+
func DecodeLabelsFromJSON(data []byte) (*LabelsMessage, error) {
234
+
var msg LabelsMessage
235
+
if err := json.Unmarshal(data, &msg); err != nil {
236
+
return nil, err
237
+
}
238
+
return &msg, nil
239
+
}
+21
pkg/appview/middleware/registry.go
+21
pkg/appview/middleware/registry.go
···
166
166
return serviceToken, err
167
167
}
168
168
169
+
// LabelChecker checks whether content has been taken down via ATProto labels.
170
+
type LabelChecker interface {
171
+
IsTakenDown(did, repository string) (bool, error)
172
+
}
173
+
169
174
// Global variables for initialization only
170
175
// These are set by main.go during startup and copied into NamespaceResolver instances.
171
176
// After initialization, request handling uses the NamespaceResolver's instance fields.
···
175
180
globalAuthorizer auth.HoldAuthorizer
176
181
globalWebhookDispatcher storage.PushWebhookDispatcher
177
182
globalManifestRefChecker storage.ManifestReferenceChecker
183
+
globalLabelChecker LabelChecker
178
184
)
179
185
180
186
// SetGlobalRefresher sets the OAuth refresher instance during initialization
···
192
198
// SetGlobalManifestRefChecker sets the manifest reference checker during initialization
193
199
func SetGlobalManifestRefChecker(checker storage.ManifestReferenceChecker) {
194
200
globalManifestRefChecker = checker
201
+
}
202
+
203
+
// SetGlobalLabelChecker sets the label checker instance during initialization
204
+
func SetGlobalLabelChecker(checker LabelChecker) {
205
+
globalLabelChecker = checker
195
206
}
196
207
197
208
// SetGlobalAuthorizer sets the authorizer instance during initialization
···
303
314
}
304
315
305
316
slog.Debug("Resolved identity", "component", "registry/middleware", "did", did, "pds", pdsEndpoint, "handle", handle)
317
+
318
+
// Check for takedown labels before proceeding
319
+
if globalLabelChecker != nil {
320
+
if taken, _ := globalLabelChecker.IsTakenDown(did, imageName); taken {
321
+
return nil, errcode.Error{
322
+
Code: errcode.ErrorCodeDenied,
323
+
Message: "this repository has been removed due to a policy violation",
324
+
}
325
+
}
326
+
}
306
327
307
328
// Query for hold DID - either user's hold or default hold service
308
329
// Also returns the sailor profile so we can read preferences (e.g. AutoRemoveUntagged)
+13
pkg/appview/server.go
+13
pkg/appview/server.go
···
24
24
"atcr.io/pkg/appview/db"
25
25
"atcr.io/pkg/appview/holdhealth"
26
26
"atcr.io/pkg/appview/jetstream"
27
+
appviewlabeler "atcr.io/pkg/appview/labeler"
27
28
"atcr.io/pkg/appview/middleware"
28
29
"atcr.io/pkg/appview/readme"
29
30
"atcr.io/pkg/appview/routes"
···
236
237
middleware.SetGlobalDatabase(holdDIDDB)
237
238
middleware.SetGlobalManifestRefChecker(holdDIDDB)
238
239
240
+
// Set label checker for takedown filtering
241
+
middleware.SetGlobalLabelChecker(db.NewLabelChecker(s.Database))
242
+
239
243
// Create RemoteHoldAuthorizer for hold authorization with caching
240
244
s.HoldAuthorizer = auth.NewRemoteHoldAuthorizer(s.Database, testMode)
241
245
middleware.SetGlobalAuthorizer(s.HoldAuthorizer)
···
286
290
287
291
// Initialize Jetstream workers
288
292
s.initializeJetstream()
293
+
294
+
// Initialize labeler subscriber
295
+
if cfg.Labeler.DID != "" {
296
+
sub := appviewlabeler.SubscriberFromConfig(cfg.Labeler.DID, s.Database)
297
+
if sub != nil {
298
+
sub.Start()
299
+
slog.Info("Labeler subscriber started", "labeler", cfg.Labeler.DID)
300
+
}
301
+
}
289
302
290
303
// Create main chi router
291
304
mainRouter := chi.NewRouter()
+106
pkg/labeler/auth.go
+106
pkg/labeler/auth.go
···
1
+
package labeler
2
+
3
+
import (
4
+
"crypto/rand"
5
+
"encoding/base64"
6
+
"net/http"
7
+
"sync"
8
+
)
9
+
10
+
// Session represents an authenticated admin session.
11
+
type Session struct {
12
+
DID string
13
+
Handle string
14
+
}
15
+
16
+
// Auth manages admin authentication.
17
+
type Auth struct {
18
+
ownerDID string
19
+
sessions map[string]*Session
20
+
sessionsMu sync.RWMutex
21
+
}
22
+
23
+
// NewAuth creates a new Auth manager.
24
+
func NewAuth(ownerDID string) *Auth {
25
+
return &Auth{
26
+
ownerDID: ownerDID,
27
+
sessions: make(map[string]*Session),
28
+
}
29
+
}
30
+
31
+
func (a *Auth) createSession(did, handle string) (string, error) {
32
+
b := make([]byte, 32)
33
+
if _, err := rand.Read(b); err != nil {
34
+
return "", err
35
+
}
36
+
token := base64.URLEncoding.EncodeToString(b)
37
+
38
+
a.sessionsMu.Lock()
39
+
a.sessions[token] = &Session{DID: did, Handle: handle}
40
+
a.sessionsMu.Unlock()
41
+
42
+
return token, nil
43
+
}
44
+
45
+
func (a *Auth) getSession(token string) *Session {
46
+
a.sessionsMu.RLock()
47
+
defer a.sessionsMu.RUnlock()
48
+
return a.sessions[token]
49
+
}
50
+
51
+
func (a *Auth) deleteSession(token string) {
52
+
a.sessionsMu.Lock()
53
+
delete(a.sessions, token)
54
+
a.sessionsMu.Unlock()
55
+
}
56
+
57
+
const sessionCookieName = "labeler_session"
58
+
59
+
func setSessionCookie(w http.ResponseWriter, r *http.Request, token string) {
60
+
secure := r.TLS != nil || r.Header.Get("X-Forwarded-Proto") == "https"
61
+
http.SetCookie(w, &http.Cookie{
62
+
Name: sessionCookieName,
63
+
Value: token,
64
+
Path: "/",
65
+
MaxAge: 86400, // 24 hours
66
+
HttpOnly: true,
67
+
Secure: secure,
68
+
SameSite: http.SameSiteLaxMode,
69
+
})
70
+
}
71
+
72
+
func clearSessionCookie(w http.ResponseWriter) {
73
+
http.SetCookie(w, &http.Cookie{
74
+
Name: sessionCookieName,
75
+
Value: "",
76
+
Path: "/",
77
+
MaxAge: -1,
78
+
HttpOnly: true,
79
+
SameSite: http.SameSiteLaxMode,
80
+
})
81
+
}
82
+
83
+
func getSessionCookie(r *http.Request) (string, bool) {
84
+
cookie, err := r.Cookie(sessionCookieName)
85
+
if err != nil {
86
+
return "", false
87
+
}
88
+
return cookie.Value, true
89
+
}
90
+
91
+
// RequireOwner is middleware that checks the session belongs to the owner DID.
92
+
func (a *Auth) RequireOwner(next http.Handler) http.Handler {
93
+
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
94
+
token, ok := getSessionCookie(r)
95
+
if !ok {
96
+
http.Redirect(w, r, "/auth/login", http.StatusFound)
97
+
return
98
+
}
99
+
session := a.getSession(token)
100
+
if session == nil || session.DID != a.ownerDID {
101
+
http.Redirect(w, r, "/auth/login", http.StatusFound)
102
+
return
103
+
}
104
+
next.ServeHTTP(w, r)
105
+
})
106
+
}
+146
pkg/labeler/config.go
+146
pkg/labeler/config.go
···
1
+
// Package labeler implements the ATCR labeler service, an ATProto-compatible
2
+
// content moderation service for issuing takedown labels on container registry content.
3
+
package labeler
4
+
5
+
import (
6
+
"fmt"
7
+
"net/url"
8
+
"strings"
9
+
10
+
"github.com/spf13/viper"
11
+
12
+
"atcr.io/pkg/config"
13
+
)
14
+
15
+
// Config represents the labeler service configuration.
16
+
// It reuses the appview config YAML structure, reading from the "labeler" section.
17
+
type Config struct {
18
+
Version string `yaml:"version" comment:"Configuration format version."`
19
+
LogLevel string `yaml:"log_level" comment:"Log level: debug, info, warn, error."`
20
+
Labeler LabelerConfig `yaml:"labeler" comment:"Labeler service settings."`
21
+
Server AppviewServerConfig `yaml:"server" comment:"AppView server settings (shared config)."`
22
+
LogShipper config.LogShipperConfig `yaml:"log_shipper" comment:"Remote log shipping settings."`
23
+
}
24
+
25
+
// LabelerConfig defines labeler-specific settings.
26
+
type LabelerConfig struct {
27
+
// Enable the labeler service.
28
+
Enabled bool `yaml:"enabled" comment:"Enable the labeler service."`
29
+
30
+
// Listen address for the labeler HTTP server.
31
+
Addr string `yaml:"addr" comment:"Listen address for labeler (e.g., :5002)."`
32
+
33
+
// DID of the labeler admin. Only this DID can log into the admin panel.
34
+
OwnerDID string `yaml:"owner_did" comment:"DID of the labeler admin. Only this DID can log into the admin panel."`
35
+
36
+
// Path to labeler SQLite database.
37
+
DBPath string `yaml:"db_path" comment:"Path to labeler SQLite database."`
38
+
}
39
+
40
+
// AppviewServerConfig is a subset of the appview ServerConfig that the labeler needs.
41
+
type AppviewServerConfig struct {
42
+
BaseURL string `yaml:"base_url"`
43
+
ClientName string `yaml:"client_name"`
44
+
ClientShortName string `yaml:"client_short_name"`
45
+
TestMode bool `yaml:"test_mode"`
46
+
}
47
+
48
+
// PublicURL returns the labeler's public URL derived from the appview base URL.
49
+
// If appview is https://atcr.io, labeler is https://labeler.atcr.io.
50
+
func (c *Config) PublicURL() string {
51
+
u, err := url.Parse(c.Server.BaseURL)
52
+
if err != nil {
53
+
return ""
54
+
}
55
+
u.Host = "labeler." + u.Host
56
+
return u.String()
57
+
}
58
+
59
+
// DID returns the labeler's did:web identity derived from its public URL.
60
+
func (c *Config) DID() string {
61
+
u, err := url.Parse(c.PublicURL())
62
+
if err != nil {
63
+
return ""
64
+
}
65
+
host := u.Hostname()
66
+
if port := u.Port(); port != "" {
67
+
host += "%3A" + port
68
+
}
69
+
return "did:web:" + host
70
+
}
71
+
72
+
func setDefaults(v *viper.Viper) {
73
+
v.SetDefault("version", "0.1")
74
+
v.SetDefault("log_level", "info")
75
+
76
+
// Labeler defaults
77
+
v.SetDefault("labeler.enabled", false)
78
+
v.SetDefault("labeler.addr", ":5002")
79
+
v.SetDefault("labeler.owner_did", "")
80
+
v.SetDefault("labeler.db_path", "/var/lib/atcr-labeler/labeler.db")
81
+
82
+
// Server defaults (read from shared appview config)
83
+
v.SetDefault("server.base_url", "")
84
+
v.SetDefault("server.client_name", "AT Container Registry")
85
+
v.SetDefault("server.client_short_name", "ATCR")
86
+
v.SetDefault("server.test_mode", false)
87
+
}
88
+
89
+
// LoadConfig loads the labeler configuration from the appview config YAML.
90
+
func LoadConfig(yamlPath string) (*Config, error) {
91
+
v := config.NewViper("LABELER", yamlPath)
92
+
setDefaults(v)
93
+
94
+
cfg := &Config{}
95
+
if err := v.Unmarshal(cfg, config.UnmarshalOption()); err != nil {
96
+
return nil, fmt.Errorf("failed to unmarshal config: %w", err)
97
+
}
98
+
99
+
// Also try ATCR_ prefix for shared server config
100
+
atcrV := config.NewViper("ATCR", yamlPath)
101
+
if baseURL := atcrV.GetString("server.base_url"); baseURL != "" && cfg.Server.BaseURL == "" {
102
+
cfg.Server.BaseURL = baseURL
103
+
}
104
+
if clientName := atcrV.GetString("server.client_name"); clientName != "" && cfg.Server.ClientName == "" {
105
+
cfg.Server.ClientName = clientName
106
+
}
107
+
if clientShortName := atcrV.GetString("server.client_short_name"); clientShortName != "" && cfg.Server.ClientShortName == "" {
108
+
cfg.Server.ClientShortName = clientShortName
109
+
}
110
+
if atcrV.GetBool("server.test_mode") {
111
+
cfg.Server.TestMode = true
112
+
}
113
+
114
+
// Validation
115
+
if cfg.Server.BaseURL == "" {
116
+
return nil, fmt.Errorf("server.base_url is required")
117
+
}
118
+
if cfg.Labeler.OwnerDID == "" {
119
+
return nil, fmt.Errorf("labeler.owner_did is required")
120
+
}
121
+
if !strings.HasPrefix(cfg.Labeler.OwnerDID, "did:") {
122
+
return nil, fmt.Errorf("labeler.owner_did must be a DID (got %q)", cfg.Labeler.OwnerDID)
123
+
}
124
+
125
+
return cfg, nil
126
+
}
127
+
128
+
// ExampleYAML generates an example labeler configuration file.
129
+
func ExampleYAML() ([]byte, error) {
130
+
cfg := &Config{
131
+
Version: "0.1",
132
+
LogLevel: "info",
133
+
Server: AppviewServerConfig{
134
+
BaseURL: "https://atcr.io",
135
+
ClientName: "AT Container Registry",
136
+
ClientShortName: "ATCR",
137
+
},
138
+
Labeler: LabelerConfig{
139
+
Enabled: true,
140
+
Addr: ":5002",
141
+
OwnerDID: "did:plc:your-did-here",
142
+
DBPath: "/var/lib/atcr-labeler/labeler.db",
143
+
},
144
+
}
145
+
return config.MarshalCommentedYAML("ATCR Labeler Configuration", cfg)
146
+
}
+51
pkg/labeler/config_test.go
+51
pkg/labeler/config_test.go
···
1
+
package labeler
2
+
3
+
import "testing"
4
+
5
+
func TestConfig_PublicURL(t *testing.T) {
6
+
tests := []struct {
7
+
name string
8
+
baseURL string
9
+
want string
10
+
}{
11
+
{"standard", "https://atcr.io", "https://labeler.atcr.io"},
12
+
{"with port", "https://atcr.io:8080", "https://labeler.atcr.io:8080"},
13
+
{"localhost", "http://localhost:5000", "http://labeler.localhost:5000"},
14
+
}
15
+
16
+
for _, tt := range tests {
17
+
t.Run(tt.name, func(t *testing.T) {
18
+
cfg := &Config{
19
+
Server: AppviewServerConfig{BaseURL: tt.baseURL},
20
+
}
21
+
got := cfg.PublicURL()
22
+
if got != tt.want {
23
+
t.Errorf("PublicURL() = %q, want %q", got, tt.want)
24
+
}
25
+
})
26
+
}
27
+
}
28
+
29
+
func TestConfig_DID(t *testing.T) {
30
+
tests := []struct {
31
+
name string
32
+
baseURL string
33
+
want string
34
+
}{
35
+
{"standard", "https://atcr.io", "did:web:labeler.atcr.io"},
36
+
{"with port", "https://atcr.io:8080", "did:web:labeler.atcr.io%3A8080"},
37
+
{"localhost", "http://localhost:5000", "did:web:labeler.localhost%3A5000"},
38
+
}
39
+
40
+
for _, tt := range tests {
41
+
t.Run(tt.name, func(t *testing.T) {
42
+
cfg := &Config{
43
+
Server: AppviewServerConfig{BaseURL: tt.baseURL},
44
+
}
45
+
got := cfg.DID()
46
+
if got != tt.want {
47
+
t.Errorf("DID() = %q, want %q", got, tt.want)
48
+
}
49
+
})
50
+
}
51
+
}
+301
pkg/labeler/db.go
+301
pkg/labeler/db.go
···
1
+
package labeler
2
+
3
+
import (
4
+
"database/sql"
5
+
"fmt"
6
+
"os"
7
+
"path/filepath"
8
+
"time"
9
+
10
+
_ "github.com/tursodatabase/go-libsql"
11
+
)
12
+
13
+
const schema = `
14
+
CREATE TABLE IF NOT EXISTS labels (
15
+
id INTEGER PRIMARY KEY AUTOINCREMENT,
16
+
src TEXT NOT NULL,
17
+
uri TEXT NOT NULL,
18
+
cid TEXT,
19
+
val TEXT NOT NULL,
20
+
neg BOOLEAN NOT NULL DEFAULT 0,
21
+
cts TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
22
+
exp TIMESTAMP,
23
+
subject_did TEXT NOT NULL,
24
+
subject_repo TEXT NOT NULL DEFAULT '',
25
+
UNIQUE(src, uri, val, neg)
26
+
);
27
+
CREATE INDEX IF NOT EXISTS idx_labels_subject ON labels(subject_did, subject_repo);
28
+
CREATE INDEX IF NOT EXISTS idx_labels_cts ON labels(cts DESC);
29
+
`
30
+
31
+
// Label represents an ATProto label (com.atproto.label.defs#label).
32
+
type Label struct {
33
+
ID int64
34
+
Src string
35
+
URI string
36
+
CID string
37
+
Val string
38
+
Neg bool
39
+
Cts time.Time
40
+
Exp *time.Time
41
+
SubjectDID string
42
+
SubjectRepo string
43
+
}
44
+
45
+
// OpenDB opens or creates the labeler database.
46
+
func OpenDB(dbPath string) (*sql.DB, error) {
47
+
if err := os.MkdirAll(filepath.Dir(dbPath), 0755); err != nil {
48
+
return nil, fmt.Errorf("failed to create db directory: %w", err)
49
+
}
50
+
51
+
db, err := sql.Open("libsql", "file:"+dbPath)
52
+
if err != nil {
53
+
return nil, fmt.Errorf("failed to open database: %w", err)
54
+
}
55
+
56
+
// Apply schema
57
+
for _, stmt := range splitStatements(schema) {
58
+
if _, err := db.Exec(stmt); err != nil {
59
+
return nil, fmt.Errorf("failed to apply schema: %w", err)
60
+
}
61
+
}
62
+
63
+
return db, nil
64
+
}
65
+
66
+
// splitStatements splits SQL by semicolons (go-libsql doesn't support multi-statement exec).
67
+
func splitStatements(sql string) []string {
68
+
var stmts []string
69
+
for _, s := range splitOnSemicolon(sql) {
70
+
s = trimSpace(s)
71
+
if s != "" {
72
+
stmts = append(stmts, s)
73
+
}
74
+
}
75
+
return stmts
76
+
}
77
+
78
+
func splitOnSemicolon(s string) []string {
79
+
var parts []string
80
+
start := 0
81
+
for i := 0; i < len(s); i++ {
82
+
if s[i] == ';' {
83
+
parts = append(parts, s[start:i])
84
+
start = i + 1
85
+
}
86
+
}
87
+
if start < len(s) {
88
+
parts = append(parts, s[start:])
89
+
}
90
+
return parts
91
+
}
92
+
93
+
func trimSpace(s string) string {
94
+
// Simple trim that handles newlines and spaces
95
+
i := 0
96
+
for i < len(s) && (s[i] == ' ' || s[i] == '\t' || s[i] == '\n' || s[i] == '\r') {
97
+
i++
98
+
}
99
+
j := len(s)
100
+
for j > i && (s[j-1] == ' ' || s[j-1] == '\t' || s[j-1] == '\n' || s[j-1] == '\r') {
101
+
j--
102
+
}
103
+
return s[i:j]
104
+
}
105
+
106
+
// CreateLabel inserts a new label into the database.
107
+
func CreateLabel(db *sql.DB, l *Label) (int64, error) {
108
+
result, err := db.Exec(
109
+
`INSERT INTO labels (src, uri, cid, val, neg, cts, exp, subject_did, subject_repo)
110
+
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
111
+
ON CONFLICT(src, uri, val, neg) DO UPDATE SET cts = excluded.cts`,
112
+
l.Src, l.URI, l.CID, l.Val, l.Neg, l.Cts.UTC().Format(time.RFC3339), l.Exp,
113
+
l.SubjectDID, l.SubjectRepo,
114
+
)
115
+
if err != nil {
116
+
return 0, fmt.Errorf("failed to create label: %w", err)
117
+
}
118
+
return result.LastInsertId()
119
+
}
120
+
121
+
// NegateLabel creates a negation label to reverse a previous label.
122
+
func NegateLabel(db *sql.DB, src, uri, val string, subjectDID, subjectRepo string) error {
123
+
_, err := db.Exec(
124
+
`INSERT INTO labels (src, uri, val, neg, cts, subject_did, subject_repo)
125
+
VALUES (?, ?, ?, 1, ?, ?, ?)`,
126
+
src, uri, val, time.Now().UTC().Format(time.RFC3339), subjectDID, subjectRepo,
127
+
)
128
+
return err
129
+
}
130
+
131
+
// GetLabelsSince returns labels with id > cursor, ordered by id ascending.
132
+
func GetLabelsSince(db *sql.DB, cursor int64, limit int) ([]Label, error) {
133
+
rows, err := db.Query(
134
+
`SELECT id, src, uri, COALESCE(cid, ''), val, neg, cts, exp, subject_did, subject_repo
135
+
FROM labels WHERE id > ? ORDER BY id ASC LIMIT ?`,
136
+
cursor, limit,
137
+
)
138
+
if err != nil {
139
+
return nil, err
140
+
}
141
+
defer rows.Close()
142
+
143
+
return scanLabels(rows)
144
+
}
145
+
146
+
// ListActiveTakedowns returns active (non-negated) takedown labels.
147
+
func ListActiveTakedowns(db *sql.DB, limit, offset int) ([]Label, int, error) {
148
+
var total int
149
+
err := db.QueryRow(
150
+
`SELECT COUNT(*) FROM labels l1
151
+
WHERE l1.val = '!takedown' AND l1.neg = 0
152
+
AND NOT EXISTS (
153
+
SELECT 1 FROM labels l2
154
+
WHERE l2.src = l1.src AND l2.uri = l1.uri AND l2.val = l1.val
155
+
AND l2.neg = 1 AND l2.id > l1.id
156
+
)
157
+
AND (l1.exp IS NULL OR l1.exp > CURRENT_TIMESTAMP)`,
158
+
).Scan(&total)
159
+
if err != nil {
160
+
return nil, 0, err
161
+
}
162
+
163
+
rows, err := db.Query(
164
+
`SELECT l1.id, l1.src, l1.uri, COALESCE(l1.cid, ''), l1.val, l1.neg, l1.cts, l1.exp, l1.subject_did, l1.subject_repo
165
+
FROM labels l1
166
+
WHERE l1.val = '!takedown' AND l1.neg = 0
167
+
AND NOT EXISTS (
168
+
SELECT 1 FROM labels l2
169
+
WHERE l2.src = l1.src AND l2.uri = l1.uri AND l2.val = l1.val
170
+
AND l2.neg = 1 AND l2.id > l1.id
171
+
)
172
+
AND (l1.exp IS NULL OR l1.exp > CURRENT_TIMESTAMP)
173
+
ORDER BY l1.cts DESC LIMIT ? OFFSET ?`,
174
+
limit, offset,
175
+
)
176
+
if err != nil {
177
+
return nil, 0, err
178
+
}
179
+
defer rows.Close()
180
+
181
+
labels, err := scanLabels(rows)
182
+
return labels, total, err
183
+
}
184
+
185
+
// GetLabelsForRepo returns all active labels for a specific DID + repository.
186
+
func GetLabelsForRepo(db *sql.DB, did, repo string) ([]Label, error) {
187
+
rows, err := db.Query(
188
+
`SELECT id, src, uri, COALESCE(cid, ''), val, neg, cts, exp, subject_did, subject_repo
189
+
FROM labels
190
+
WHERE subject_did = ? AND subject_repo = ?
191
+
ORDER BY cts DESC`,
192
+
did, repo,
193
+
)
194
+
if err != nil {
195
+
return nil, err
196
+
}
197
+
defer rows.Close()
198
+
return scanLabels(rows)
199
+
}
200
+
201
+
// NegateRepoLabels creates negation labels for all active takedown labels on a (DID, repo) pair.
202
+
func NegateRepoLabels(db *sql.DB, src, did, repo string) error {
203
+
rows, err := db.Query(
204
+
`SELECT uri FROM labels
205
+
WHERE subject_did = ? AND subject_repo = ? AND val = '!takedown' AND neg = 0`,
206
+
did, repo,
207
+
)
208
+
if err != nil {
209
+
return err
210
+
}
211
+
212
+
var uris []string
213
+
for rows.Next() {
214
+
var uri string
215
+
if err := rows.Scan(&uri); err != nil {
216
+
rows.Close()
217
+
return err
218
+
}
219
+
uris = append(uris, uri)
220
+
}
221
+
rows.Close()
222
+
if err := rows.Err(); err != nil {
223
+
return err
224
+
}
225
+
226
+
now := time.Now().UTC().Format(time.RFC3339)
227
+
for _, uri := range uris {
228
+
if _, err := db.Exec(
229
+
`INSERT INTO labels (src, uri, val, neg, cts, subject_did, subject_repo)
230
+
VALUES (?, ?, '!takedown', 1, ?, ?, ?)`,
231
+
src, uri, now, did, repo,
232
+
); err != nil {
233
+
return err
234
+
}
235
+
}
236
+
return nil
237
+
}
238
+
239
+
// NegateUserLabels creates negation labels for all active takedown labels on a DID (user-level).
240
+
func NegateUserLabels(db *sql.DB, src, did string) error {
241
+
rows, err := db.Query(
242
+
`SELECT uri, subject_repo FROM labels
243
+
WHERE subject_did = ? AND val = '!takedown' AND neg = 0`,
244
+
did,
245
+
)
246
+
if err != nil {
247
+
return err
248
+
}
249
+
250
+
type uriRepo struct {
251
+
uri string
252
+
repo string
253
+
}
254
+
var entries []uriRepo
255
+
for rows.Next() {
256
+
var e uriRepo
257
+
if err := rows.Scan(&e.uri, &e.repo); err != nil {
258
+
rows.Close()
259
+
return err
260
+
}
261
+
entries = append(entries, e)
262
+
}
263
+
rows.Close()
264
+
if err := rows.Err(); err != nil {
265
+
return err
266
+
}
267
+
268
+
now := time.Now().UTC().Format(time.RFC3339)
269
+
for _, e := range entries {
270
+
if _, err := db.Exec(
271
+
`INSERT INTO labels (src, uri, val, neg, cts, subject_did, subject_repo)
272
+
VALUES (?, ?, '!takedown', 1, ?, ?, ?)`,
273
+
src, e.uri, now, did, e.repo,
274
+
); err != nil {
275
+
return err
276
+
}
277
+
}
278
+
return nil
279
+
}
280
+
281
+
func scanLabels(rows *sql.Rows) ([]Label, error) {
282
+
var labels []Label
283
+
for rows.Next() {
284
+
var l Label
285
+
var cts string
286
+
var exp *string
287
+
if err := rows.Scan(&l.ID, &l.Src, &l.URI, &l.CID, &l.Val, &l.Neg, &cts, &exp, &l.SubjectDID, &l.SubjectRepo); err != nil {
288
+
return nil, err
289
+
}
290
+
if t, err := time.Parse(time.RFC3339, cts); err == nil {
291
+
l.Cts = t
292
+
}
293
+
if exp != nil {
294
+
if t, err := time.Parse(time.RFC3339, *exp); err == nil {
295
+
l.Exp = &t
296
+
}
297
+
}
298
+
labels = append(labels, l)
299
+
}
300
+
return labels, rows.Err()
301
+
}
+412
pkg/labeler/db_test.go
+412
pkg/labeler/db_test.go
···
1
+
package labeler
2
+
3
+
import (
4
+
"os"
5
+
"path/filepath"
6
+
"testing"
7
+
"time"
8
+
)
9
+
10
+
func TestOpenDB(t *testing.T) {
11
+
dir := t.TempDir()
12
+
dbPath := filepath.Join(dir, "subdir", "test.db")
13
+
14
+
db, err := OpenDB(dbPath)
15
+
if err != nil {
16
+
t.Fatalf("OpenDB failed: %v", err)
17
+
}
18
+
defer db.Close()
19
+
20
+
// Verify directory was created
21
+
if _, err := os.Stat(filepath.Dir(dbPath)); os.IsNotExist(err) {
22
+
t.Error("expected directory to be created")
23
+
}
24
+
25
+
// Verify tables exist
26
+
var count int
27
+
err = db.QueryRow("SELECT COUNT(*) FROM labels").Scan(&count)
28
+
if err != nil {
29
+
t.Fatalf("failed to query labels table: %v", err)
30
+
}
31
+
if count != 0 {
32
+
t.Errorf("expected 0 labels, got %d", count)
33
+
}
34
+
}
35
+
36
+
func TestCreateLabel(t *testing.T) {
37
+
dir := t.TempDir()
38
+
db, err := OpenDB(filepath.Join(dir, "test.db"))
39
+
if err != nil {
40
+
t.Fatal(err)
41
+
}
42
+
defer db.Close()
43
+
44
+
now := time.Now().UTC().Truncate(time.Second)
45
+
label := &Label{
46
+
Src: "did:web:labeler.atcr.io",
47
+
URI: "at://did:plc:abc/io.atcr.manifest/sha256-123",
48
+
Val: "!takedown",
49
+
Cts: now,
50
+
SubjectDID: "did:plc:abc",
51
+
SubjectRepo: "myimage",
52
+
}
53
+
54
+
id, err := CreateLabel(db, label)
55
+
if err != nil {
56
+
t.Fatalf("CreateLabel failed: %v", err)
57
+
}
58
+
if id <= 0 {
59
+
t.Errorf("expected positive id, got %d", id)
60
+
}
61
+
62
+
// Verify it was stored
63
+
labels, err := GetLabelsSince(db, 0, 10)
64
+
if err != nil {
65
+
t.Fatal(err)
66
+
}
67
+
if len(labels) != 1 {
68
+
t.Fatalf("expected 1 label, got %d", len(labels))
69
+
}
70
+
if labels[0].Src != "did:web:labeler.atcr.io" {
71
+
t.Errorf("expected src did:web:labeler.atcr.io, got %s", labels[0].Src)
72
+
}
73
+
if labels[0].Val != "!takedown" {
74
+
t.Errorf("expected val !takedown, got %s", labels[0].Val)
75
+
}
76
+
if labels[0].SubjectDID != "did:plc:abc" {
77
+
t.Errorf("expected subject_did did:plc:abc, got %s", labels[0].SubjectDID)
78
+
}
79
+
if labels[0].SubjectRepo != "myimage" {
80
+
t.Errorf("expected subject_repo myimage, got %s", labels[0].SubjectRepo)
81
+
}
82
+
}
83
+
84
+
func TestCreateLabel_Upsert(t *testing.T) {
85
+
dir := t.TempDir()
86
+
db, err := OpenDB(filepath.Join(dir, "test.db"))
87
+
if err != nil {
88
+
t.Fatal(err)
89
+
}
90
+
defer db.Close()
91
+
92
+
now := time.Now().UTC()
93
+
label := &Label{
94
+
Src: "did:web:labeler.atcr.io",
95
+
URI: "at://did:plc:abc/io.atcr.manifest/sha256-123",
96
+
Val: "!takedown",
97
+
Cts: now,
98
+
SubjectDID: "did:plc:abc",
99
+
SubjectRepo: "myimage",
100
+
}
101
+
102
+
// First insert
103
+
_, err = CreateLabel(db, label)
104
+
if err != nil {
105
+
t.Fatal(err)
106
+
}
107
+
108
+
// Same (src, uri, val) - should upsert, not error
109
+
label.Cts = now.Add(time.Hour)
110
+
_, err = CreateLabel(db, label)
111
+
if err != nil {
112
+
t.Fatalf("upsert should not fail: %v", err)
113
+
}
114
+
115
+
// Should still be 1 label
116
+
labels, err := GetLabelsSince(db, 0, 10)
117
+
if err != nil {
118
+
t.Fatal(err)
119
+
}
120
+
if len(labels) != 1 {
121
+
t.Errorf("expected 1 label after upsert, got %d", len(labels))
122
+
}
123
+
}
124
+
125
+
func TestNegateLabel(t *testing.T) {
126
+
dir := t.TempDir()
127
+
db, err := OpenDB(filepath.Join(dir, "test.db"))
128
+
if err != nil {
129
+
t.Fatal(err)
130
+
}
131
+
defer db.Close()
132
+
133
+
src := "did:web:labeler.atcr.io"
134
+
now := time.Now().UTC()
135
+
136
+
// Create a label
137
+
_, err = CreateLabel(db, &Label{
138
+
Src: src, URI: "at://did:plc:abc/io.atcr.manifest/sha256-123",
139
+
Val: "!takedown", Cts: now,
140
+
SubjectDID: "did:plc:abc", SubjectRepo: "myimage",
141
+
})
142
+
if err != nil {
143
+
t.Fatal(err)
144
+
}
145
+
146
+
// Negate it
147
+
err = NegateLabel(db, src, "at://did:plc:abc/io.atcr.manifest/sha256-123", "!takedown", "did:plc:abc", "myimage")
148
+
if err != nil {
149
+
t.Fatalf("NegateLabel failed: %v", err)
150
+
}
151
+
152
+
// Should have 2 labels now (original + negation)
153
+
labels, err := GetLabelsSince(db, 0, 10)
154
+
if err != nil {
155
+
t.Fatal(err)
156
+
}
157
+
if len(labels) != 2 {
158
+
t.Fatalf("expected 2 labels, got %d", len(labels))
159
+
}
160
+
161
+
// The negation label should have neg=true
162
+
negLabel := labels[1]
163
+
if !negLabel.Neg {
164
+
t.Error("expected negation label to have neg=true")
165
+
}
166
+
}
167
+
168
+
func TestListActiveTakedowns(t *testing.T) {
169
+
dir := t.TempDir()
170
+
db, err := OpenDB(filepath.Join(dir, "test.db"))
171
+
if err != nil {
172
+
t.Fatal(err)
173
+
}
174
+
defer db.Close()
175
+
176
+
src := "did:web:labeler.atcr.io"
177
+
now := time.Now().UTC()
178
+
179
+
// Create 3 labels
180
+
for i, repo := range []string{"repo1", "repo2", "repo3"} {
181
+
_, err = CreateLabel(db, &Label{
182
+
Src: src, URI: "at://did:plc:abc/io.atcr.repo/" + repo,
183
+
Val: "!takedown", Cts: now.Add(time.Duration(i) * time.Minute),
184
+
SubjectDID: "did:plc:abc", SubjectRepo: repo,
185
+
})
186
+
if err != nil {
187
+
t.Fatal(err)
188
+
}
189
+
}
190
+
191
+
// All 3 should be active
192
+
labels, total, err := ListActiveTakedowns(db, 10, 0)
193
+
if err != nil {
194
+
t.Fatal(err)
195
+
}
196
+
if total != 3 {
197
+
t.Errorf("expected 3 active takedowns, got %d", total)
198
+
}
199
+
if len(labels) != 3 {
200
+
t.Errorf("expected 3 labels returned, got %d", len(labels))
201
+
}
202
+
203
+
// Negate one
204
+
err = NegateLabel(db, src, "at://did:plc:abc/io.atcr.repo/repo2", "!takedown", "did:plc:abc", "repo2")
205
+
if err != nil {
206
+
t.Fatal(err)
207
+
}
208
+
209
+
// Should be 2 active
210
+
_, total, err = ListActiveTakedowns(db, 10, 0)
211
+
if err != nil {
212
+
t.Fatal(err)
213
+
}
214
+
if total != 2 {
215
+
t.Errorf("expected 2 active takedowns after negation, got %d", total)
216
+
}
217
+
}
218
+
219
+
func TestNegateRepoLabels(t *testing.T) {
220
+
dir := t.TempDir()
221
+
db, err := OpenDB(filepath.Join(dir, "test.db"))
222
+
if err != nil {
223
+
t.Fatal(err)
224
+
}
225
+
defer db.Close()
226
+
227
+
src := "did:web:labeler.atcr.io"
228
+
now := time.Now().UTC()
229
+
did := "did:plc:abc"
230
+
231
+
// Create multiple labels for same repo
232
+
uris := []string{
233
+
"at://did:plc:abc/io.atcr.manifest/sha256-111",
234
+
"at://did:plc:abc/io.atcr.manifest/sha256-222",
235
+
"at://did:plc:abc/io.atcr.tag/myimage-latest",
236
+
}
237
+
for _, uri := range uris {
238
+
_, err = CreateLabel(db, &Label{
239
+
Src: src, URI: uri, Val: "!takedown", Cts: now,
240
+
SubjectDID: did, SubjectRepo: "myimage",
241
+
})
242
+
if err != nil {
243
+
t.Fatal(err)
244
+
}
245
+
}
246
+
247
+
// Negate all labels for the repo
248
+
err = NegateRepoLabels(db, src, did, "myimage")
249
+
if err != nil {
250
+
t.Fatal(err)
251
+
}
252
+
253
+
// Should have 0 active takedowns
254
+
_, total, err := ListActiveTakedowns(db, 10, 0)
255
+
if err != nil {
256
+
t.Fatal(err)
257
+
}
258
+
if total != 0 {
259
+
t.Errorf("expected 0 active takedowns after repo negation, got %d", total)
260
+
}
261
+
}
262
+
263
+
func TestNegateUserLabels(t *testing.T) {
264
+
dir := t.TempDir()
265
+
db, err := OpenDB(filepath.Join(dir, "test.db"))
266
+
if err != nil {
267
+
t.Fatal(err)
268
+
}
269
+
defer db.Close()
270
+
271
+
src := "did:web:labeler.atcr.io"
272
+
now := time.Now().UTC()
273
+
did := "did:plc:abc"
274
+
275
+
// Create labels for different repos + a user-level label
276
+
_, err = CreateLabel(db, &Label{
277
+
Src: src, URI: "at://did:plc:abc", Val: "!takedown", Cts: now,
278
+
SubjectDID: did, SubjectRepo: "",
279
+
})
280
+
if err != nil {
281
+
t.Fatal(err)
282
+
}
283
+
_, err = CreateLabel(db, &Label{
284
+
Src: src, URI: "at://did:plc:abc/io.atcr.repo/repo1", Val: "!takedown", Cts: now,
285
+
SubjectDID: did, SubjectRepo: "repo1",
286
+
})
287
+
if err != nil {
288
+
t.Fatal(err)
289
+
}
290
+
291
+
// Negate all labels for the user
292
+
err = NegateUserLabels(db, src, did)
293
+
if err != nil {
294
+
t.Fatal(err)
295
+
}
296
+
297
+
// Should have 0 active
298
+
_, total, err := ListActiveTakedowns(db, 10, 0)
299
+
if err != nil {
300
+
t.Fatal(err)
301
+
}
302
+
if total != 0 {
303
+
t.Errorf("expected 0 active takedowns after user negation, got %d", total)
304
+
}
305
+
}
306
+
307
+
func TestGetLabelsSince(t *testing.T) {
308
+
dir := t.TempDir()
309
+
db, err := OpenDB(filepath.Join(dir, "test.db"))
310
+
if err != nil {
311
+
t.Fatal(err)
312
+
}
313
+
defer db.Close()
314
+
315
+
src := "did:web:labeler.atcr.io"
316
+
now := time.Now().UTC()
317
+
318
+
// Create 5 labels
319
+
for i := 0; i < 5; i++ {
320
+
_, err = CreateLabel(db, &Label{
321
+
Src: src, URI: "at://did:plc:abc/io.atcr.manifest/" + string(rune('a'+i)),
322
+
Val: "!takedown", Cts: now.Add(time.Duration(i) * time.Minute),
323
+
SubjectDID: "did:plc:abc", SubjectRepo: "repo",
324
+
})
325
+
if err != nil {
326
+
t.Fatal(err)
327
+
}
328
+
}
329
+
330
+
// Get all since 0
331
+
labels, err := GetLabelsSince(db, 0, 10)
332
+
if err != nil {
333
+
t.Fatal(err)
334
+
}
335
+
if len(labels) != 5 {
336
+
t.Errorf("expected 5 labels, got %d", len(labels))
337
+
}
338
+
339
+
// Get since cursor (skip first 3)
340
+
if len(labels) >= 3 {
341
+
cursor := labels[2].ID
342
+
after, err := GetLabelsSince(db, cursor, 10)
343
+
if err != nil {
344
+
t.Fatal(err)
345
+
}
346
+
if len(after) != 2 {
347
+
t.Errorf("expected 2 labels after cursor %d, got %d", cursor, len(after))
348
+
}
349
+
}
350
+
351
+
// Get with limit
352
+
limited, err := GetLabelsSince(db, 0, 2)
353
+
if err != nil {
354
+
t.Fatal(err)
355
+
}
356
+
if len(limited) != 2 {
357
+
t.Errorf("expected 2 labels with limit, got %d", len(limited))
358
+
}
359
+
}
360
+
361
+
func TestGetLabelsForRepo(t *testing.T) {
362
+
dir := t.TempDir()
363
+
db, err := OpenDB(filepath.Join(dir, "test.db"))
364
+
if err != nil {
365
+
t.Fatal(err)
366
+
}
367
+
defer db.Close()
368
+
369
+
src := "did:web:labeler.atcr.io"
370
+
now := time.Now().UTC()
371
+
372
+
// Labels for different repos
373
+
_, _ = CreateLabel(db, &Label{
374
+
Src: src, URI: "at://did:plc:abc/io.atcr.repo/repo1",
375
+
Val: "!takedown", Cts: now, SubjectDID: "did:plc:abc", SubjectRepo: "repo1",
376
+
})
377
+
_, _ = CreateLabel(db, &Label{
378
+
Src: src, URI: "at://did:plc:abc/io.atcr.repo/repo2",
379
+
Val: "!takedown", Cts: now, SubjectDID: "did:plc:abc", SubjectRepo: "repo2",
380
+
})
381
+
_, _ = CreateLabel(db, &Label{
382
+
Src: src, URI: "at://did:plc:def/io.atcr.repo/repo1",
383
+
Val: "!takedown", Cts: now, SubjectDID: "did:plc:def", SubjectRepo: "repo1",
384
+
})
385
+
386
+
// Get labels for specific did+repo
387
+
labels, err := GetLabelsForRepo(db, "did:plc:abc", "repo1")
388
+
if err != nil {
389
+
t.Fatal(err)
390
+
}
391
+
if len(labels) != 1 {
392
+
t.Errorf("expected 1 label for did:plc:abc/repo1, got %d", len(labels))
393
+
}
394
+
395
+
// Different user same repo
396
+
labels, err = GetLabelsForRepo(db, "did:plc:def", "repo1")
397
+
if err != nil {
398
+
t.Fatal(err)
399
+
}
400
+
if len(labels) != 1 {
401
+
t.Errorf("expected 1 label for did:plc:def/repo1, got %d", len(labels))
402
+
}
403
+
404
+
// No labels
405
+
labels, err = GetLabelsForRepo(db, "did:plc:xyz", "repo1")
406
+
if err != nil {
407
+
t.Fatal(err)
408
+
}
409
+
if len(labels) != 0 {
410
+
t.Errorf("expected 0 labels for unknown did, got %d", len(labels))
411
+
}
412
+
}
+118
pkg/labeler/handlers.go
+118
pkg/labeler/handlers.go
···
1
+
package labeler
2
+
3
+
import (
4
+
"fmt"
5
+
"html/template"
6
+
"log/slog"
7
+
"net/http"
8
+
"strings"
9
+
10
+
"atcr.io/pkg/atproto"
11
+
)
12
+
13
+
// Auth handlers
14
+
15
+
func (s *Server) handleLogin(w http.ResponseWriter, r *http.Request) {
16
+
// If already logged in, redirect to dashboard
17
+
if token, ok := getSessionCookie(r); ok {
18
+
if session := s.auth.getSession(token); session != nil && session.DID == s.config.Labeler.OwnerDID {
19
+
http.Redirect(w, r, "/", http.StatusFound)
20
+
return
21
+
}
22
+
}
23
+
24
+
errorMsg := r.URL.Query().Get("error")
25
+
w.Header().Set("Content-Type", "text/html")
26
+
fmt.Fprintf(w, `<!DOCTYPE html>
27
+
<html>
28
+
<head><title>%s Labeler - Login</title>
29
+
<style>body{font-family:system-ui;max-width:400px;margin:100px auto;padding:0 20px}
30
+
.error{color:red;margin-bottom:1em}
31
+
input{width:100%%;padding:8px;margin:8px 0;box-sizing:border-box}
32
+
button{padding:10px 20px;cursor:pointer}</style>
33
+
</head>
34
+
<body>
35
+
<h1>%s Labeler</h1>
36
+
<p>Sign in with your AT Protocol identity.</p>
37
+
%s
38
+
<form action="/auth/oauth/authorize" method="GET">
39
+
<input name="handle" placeholder="your.handle.com" required>
40
+
<button type="submit">Sign In</button>
41
+
</form>
42
+
</body></html>`,
43
+
s.config.Server.ClientShortName,
44
+
s.config.Server.ClientShortName,
45
+
func() string {
46
+
if errorMsg != "" {
47
+
return fmt.Sprintf(`<div class="error">%s</div>`, template.HTMLEscapeString(errorMsg))
48
+
}
49
+
return ""
50
+
}(),
51
+
)
52
+
}
53
+
54
+
func (s *Server) handleAuthorize(w http.ResponseWriter, r *http.Request) {
55
+
handle := strings.TrimSpace(r.URL.Query().Get("handle"))
56
+
if handle == "" {
57
+
http.Redirect(w, r, "/auth/login?error=Handle+is+required", http.StatusFound)
58
+
return
59
+
}
60
+
61
+
handle = strings.TrimPrefix(handle, "@")
62
+
63
+
did, _, _, err := atproto.ResolveIdentity(r.Context(), handle)
64
+
if err != nil {
65
+
slog.Warn("Failed to resolve handle for labeler login", "handle", handle, "error", err)
66
+
http.Redirect(w, r, "/auth/login?error=Could+not+resolve+handle", http.StatusFound)
67
+
return
68
+
}
69
+
70
+
authURL, err := s.clientApp.StartAuthFlow(r.Context(), did)
71
+
if err != nil {
72
+
slog.Error("Failed to start OAuth flow", "error", err)
73
+
http.Redirect(w, r, "/auth/login?error=OAuth+initialization+failed", http.StatusFound)
74
+
return
75
+
}
76
+
77
+
http.Redirect(w, r, authURL, http.StatusFound)
78
+
}
79
+
80
+
func (s *Server) handleCallback(w http.ResponseWriter, r *http.Request) {
81
+
sessionData, err := s.clientApp.ProcessCallback(r.Context(), r.URL.Query())
82
+
if err != nil {
83
+
slog.Error("OAuth callback failed", "error", err)
84
+
http.Redirect(w, r, "/auth/login?error=OAuth+authentication+failed", http.StatusFound)
85
+
return
86
+
}
87
+
88
+
did := sessionData.AccountDID.String()
89
+
90
+
_, handle, _, err := atproto.ResolveIdentity(r.Context(), did)
91
+
if err != nil {
92
+
handle = did
93
+
}
94
+
95
+
// Only allow the owner
96
+
if did != s.config.Labeler.OwnerDID {
97
+
slog.Warn("Non-owner attempted labeler access", "did", did, "handle", handle, "owner", s.config.Labeler.OwnerDID)
98
+
http.Redirect(w, r, "/auth/login?error=Access+denied:+Only+the+labeler+owner+can+access+the+admin+panel", http.StatusFound)
99
+
return
100
+
}
101
+
102
+
token, err := s.auth.createSession(did, handle)
103
+
if err != nil {
104
+
http.Error(w, "Failed to create session", http.StatusInternalServerError)
105
+
return
106
+
}
107
+
setSessionCookie(w, r, token)
108
+
109
+
http.Redirect(w, r, "/", http.StatusFound)
110
+
}
111
+
112
+
func (s *Server) handleLogout(w http.ResponseWriter, r *http.Request) {
113
+
if token, ok := getSessionCookie(r); ok {
114
+
s.auth.deleteSession(token)
115
+
}
116
+
clearSessionCookie(w)
117
+
http.Redirect(w, r, "/auth/login", http.StatusFound)
118
+
}
+57
pkg/labeler/identity.go
+57
pkg/labeler/identity.go
···
1
+
package labeler
2
+
3
+
import (
4
+
"encoding/json"
5
+
"fmt"
6
+
"net/http"
7
+
)
8
+
9
+
// DIDDocument represents a did:web DID document.
10
+
type DIDDocument struct {
11
+
Context []string `json:"@context"`
12
+
ID string `json:"id"`
13
+
Service []DIDService `json:"service,omitempty"`
14
+
}
15
+
16
+
// DIDService represents a service entry in a DID document.
17
+
type DIDService struct {
18
+
ID string `json:"id"`
19
+
Type string `json:"type"`
20
+
ServiceEndpoint string `json:"serviceEndpoint"`
21
+
}
22
+
23
+
func (s *Server) handleDIDDocument(w http.ResponseWriter, r *http.Request) {
24
+
doc := DIDDocument{
25
+
Context: []string{"https://www.w3.org/ns/did/v1"},
26
+
ID: s.config.DID(),
27
+
Service: []DIDService{
28
+
{
29
+
ID: "#atproto_labeler",
30
+
Type: "AtprotoLabeler",
31
+
ServiceEndpoint: s.config.PublicURL(),
32
+
},
33
+
},
34
+
}
35
+
36
+
w.Header().Set("Content-Type", "application/json")
37
+
_ = json.NewEncoder(w).Encode(doc)
38
+
}
39
+
40
+
func (s *Server) handleClientMetadata(w http.ResponseWriter, r *http.Request) {
41
+
publicURL := s.config.PublicURL()
42
+
metadata := map[string]any{
43
+
"client_id": publicURL + "/oauth-client-metadata.json",
44
+
"client_name": fmt.Sprintf("%s Labeler", s.config.Server.ClientShortName),
45
+
"client_uri": publicURL,
46
+
"redirect_uris": []string{publicURL + "/auth/oauth/callback"},
47
+
"scope": "atproto",
48
+
"grant_types": []string{"authorization_code"},
49
+
"response_types": []string{"code"},
50
+
"token_endpoint_auth_method": "none",
51
+
"application_type": "web",
52
+
"dpop_bound_access_tokens": true,
53
+
}
54
+
55
+
w.Header().Set("Content-Type", "application/json")
56
+
_ = json.NewEncoder(w).Encode(metadata)
57
+
}
+156
pkg/labeler/server.go
+156
pkg/labeler/server.go
···
1
+
package labeler
2
+
3
+
import (
4
+
"context"
5
+
"database/sql"
6
+
"fmt"
7
+
"log/slog"
8
+
"net/http"
9
+
"net/url"
10
+
"os"
11
+
"os/signal"
12
+
"strings"
13
+
"syscall"
14
+
15
+
"atcr.io/pkg/atproto"
16
+
indigooauth "github.com/bluesky-social/indigo/atproto/auth/oauth"
17
+
"github.com/go-chi/chi/v5"
18
+
)
19
+
20
+
// Server is the labeler HTTP server.
21
+
type Server struct {
22
+
config *Config
23
+
db *sql.DB
24
+
router chi.Router
25
+
clientApp *indigooauth.ClientApp
26
+
auth *Auth
27
+
}
28
+
29
+
// NewServer creates a new labeler server.
30
+
func NewServer(cfg *Config) (*Server, error) {
31
+
db, err := OpenDB(cfg.Labeler.DBPath)
32
+
if err != nil {
33
+
return nil, fmt.Errorf("failed to open database: %w", err)
34
+
}
35
+
36
+
publicURL := cfg.PublicURL()
37
+
38
+
// Set up OAuth client for admin login
39
+
oauthStore := indigooauth.NewMemStore()
40
+
scopes := []string{"atproto"}
41
+
42
+
var oauthConfig indigooauth.ClientConfig
43
+
var redirectURI string
44
+
45
+
u, err := url.Parse(publicURL)
46
+
if err != nil {
47
+
return nil, fmt.Errorf("invalid public URL: %w", err)
48
+
}
49
+
50
+
host := u.Hostname()
51
+
if isLocalhost(host) {
52
+
port := u.Port()
53
+
if port == "" {
54
+
port = "5002"
55
+
}
56
+
oauthBaseURL := "http://127.0.0.1:" + port
57
+
redirectURI = oauthBaseURL + "/auth/oauth/callback"
58
+
oauthConfig = indigooauth.NewLocalhostConfig(redirectURI, scopes)
59
+
} else {
60
+
clientID := publicURL + "/oauth-client-metadata.json"
61
+
redirectURI = publicURL + "/auth/oauth/callback"
62
+
oauthConfig = indigooauth.NewPublicConfig(clientID, redirectURI, scopes)
63
+
}
64
+
65
+
clientApp := indigooauth.NewClientApp(&oauthConfig, oauthStore)
66
+
clientApp.Dir = atproto.GetDirectory()
67
+
68
+
auth := NewAuth(cfg.Labeler.OwnerDID)
69
+
70
+
s := &Server{
71
+
config: cfg,
72
+
db: db,
73
+
clientApp: clientApp,
74
+
auth: auth,
75
+
}
76
+
77
+
s.setupRoutes()
78
+
return s, nil
79
+
}
80
+
81
+
func (s *Server) setupRoutes() {
82
+
r := chi.NewRouter()
83
+
84
+
// DID document
85
+
r.Get("/.well-known/did.json", s.handleDIDDocument)
86
+
87
+
// OAuth client metadata
88
+
r.Get("/oauth-client-metadata.json", s.handleClientMetadata)
89
+
90
+
// Auth routes (public)
91
+
r.Get("/auth/login", s.handleLogin)
92
+
r.Get("/auth/oauth/authorize", s.handleAuthorize)
93
+
r.Get("/auth/oauth/callback", s.handleCallback)
94
+
r.Get("/auth/logout", s.handleLogout)
95
+
96
+
// XRPC endpoints (public)
97
+
r.Get("/xrpc/com.atproto.label.subscribeLabels", s.handleSubscribeLabels)
98
+
r.Get("/xrpc/com.atproto.label.queryLabels", s.handleQueryLabels)
99
+
100
+
// Protected routes (require owner)
101
+
r.Group(func(r chi.Router) {
102
+
r.Use(s.auth.RequireOwner)
103
+
104
+
r.Get("/", s.handleDashboard)
105
+
r.Get("/takedown", s.handleTakedownForm)
106
+
r.Post("/takedown", s.handleTakedownSubmit)
107
+
r.Post("/reverse", s.handleReverse)
108
+
})
109
+
110
+
s.router = r
111
+
}
112
+
113
+
// Serve starts the HTTP server with graceful shutdown.
114
+
func (s *Server) Serve() error {
115
+
slog.Info("Starting labeler service",
116
+
"addr", s.config.Labeler.Addr,
117
+
"public_url", s.config.PublicURL(),
118
+
"did", s.config.DID(),
119
+
"owner", s.config.Labeler.OwnerDID,
120
+
)
121
+
122
+
srv := &http.Server{
123
+
Addr: s.config.Labeler.Addr,
124
+
Handler: s.router,
125
+
}
126
+
127
+
// Graceful shutdown
128
+
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
129
+
defer stop()
130
+
131
+
errCh := make(chan error, 1)
132
+
go func() {
133
+
errCh <- srv.ListenAndServe()
134
+
}()
135
+
136
+
select {
137
+
case err := <-errCh:
138
+
if err != http.ErrServerClosed {
139
+
return err
140
+
}
141
+
case <-ctx.Done():
142
+
slog.Info("Shutting down labeler service")
143
+
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5000000000) // 5s
144
+
defer cancel()
145
+
if err := srv.Shutdown(shutdownCtx); err != nil {
146
+
return fmt.Errorf("shutdown error: %w", err)
147
+
}
148
+
}
149
+
150
+
s.db.Close()
151
+
return nil
152
+
}
153
+
154
+
func isLocalhost(host string) bool {
155
+
return host == "localhost" || host == "127.0.0.1" || strings.HasPrefix(host, "192.168.")
156
+
}
+187
pkg/labeler/subscribe.go
+187
pkg/labeler/subscribe.go
···
1
+
package labeler
2
+
3
+
import (
4
+
"encoding/json"
5
+
"log/slog"
6
+
"net/http"
7
+
"strconv"
8
+
"time"
9
+
10
+
"github.com/gorilla/websocket"
11
+
)
12
+
13
+
var upgrader = websocket.Upgrader{
14
+
CheckOrigin: func(r *http.Request) bool { return true },
15
+
}
16
+
17
+
// LabelsMessage is the ATProto subscribeLabels wire format.
18
+
type LabelsMessage struct {
19
+
Seq int64 `json:"seq"`
20
+
Labels []LabelOutput `json:"labels"`
21
+
}
22
+
23
+
// LabelOutput is the ATProto label format for subscribeLabels/queryLabels output.
24
+
type LabelOutput struct {
25
+
Src string `json:"src"`
26
+
URI string `json:"uri"`
27
+
CID string `json:"cid,omitempty"`
28
+
Val string `json:"val"`
29
+
Neg bool `json:"neg"`
30
+
Cts string `json:"cts"`
31
+
Exp string `json:"exp,omitempty"`
32
+
}
33
+
34
+
func labelToOutput(l Label) LabelOutput {
35
+
out := LabelOutput{
36
+
Src: l.Src,
37
+
URI: l.URI,
38
+
CID: l.CID,
39
+
Val: l.Val,
40
+
Neg: l.Neg,
41
+
Cts: l.Cts.UTC().Format(time.RFC3339),
42
+
}
43
+
if l.Exp != nil {
44
+
out.Exp = l.Exp.UTC().Format(time.RFC3339)
45
+
}
46
+
return out
47
+
}
48
+
49
+
// handleSubscribeLabels implements com.atproto.label.subscribeLabels (WebSocket).
50
+
func (s *Server) handleSubscribeLabels(w http.ResponseWriter, r *http.Request) {
51
+
cursorStr := r.URL.Query().Get("cursor")
52
+
var cursor int64
53
+
if cursorStr != "" {
54
+
var err error
55
+
cursor, err = strconv.ParseInt(cursorStr, 10, 64)
56
+
if err != nil {
57
+
http.Error(w, "invalid cursor", http.StatusBadRequest)
58
+
return
59
+
}
60
+
}
61
+
62
+
conn, err := upgrader.Upgrade(w, r, nil)
63
+
if err != nil {
64
+
slog.Error("WebSocket upgrade failed", "error", err)
65
+
return
66
+
}
67
+
defer conn.Close()
68
+
69
+
slog.Info("subscribeLabels client connected", "cursor", cursor)
70
+
71
+
// Send historical labels since cursor
72
+
labels, err := GetLabelsSince(s.db, cursor, 1000)
73
+
if err != nil {
74
+
slog.Error("Failed to get labels", "error", err)
75
+
return
76
+
}
77
+
78
+
for _, l := range labels {
79
+
msg := LabelsMessage{
80
+
Seq: l.ID,
81
+
Labels: []LabelOutput{labelToOutput(l)},
82
+
}
83
+
if err := conn.WriteJSON(msg); err != nil {
84
+
return
85
+
}
86
+
cursor = l.ID
87
+
}
88
+
89
+
// Poll for new labels
90
+
ticker := time.NewTicker(5 * time.Second)
91
+
defer ticker.Stop()
92
+
93
+
// Read pump (detect client disconnect)
94
+
done := make(chan struct{})
95
+
go func() {
96
+
defer close(done)
97
+
for {
98
+
if _, _, err := conn.ReadMessage(); err != nil {
99
+
return
100
+
}
101
+
}
102
+
}()
103
+
104
+
for {
105
+
select {
106
+
case <-done:
107
+
return
108
+
case <-ticker.C:
109
+
labels, err := GetLabelsSince(s.db, cursor, 100)
110
+
if err != nil {
111
+
slog.Error("Failed to poll labels", "error", err)
112
+
continue
113
+
}
114
+
for _, l := range labels {
115
+
msg := LabelsMessage{
116
+
Seq: l.ID,
117
+
Labels: []LabelOutput{labelToOutput(l)},
118
+
}
119
+
if err := conn.WriteJSON(msg); err != nil {
120
+
return
121
+
}
122
+
cursor = l.ID
123
+
}
124
+
}
125
+
}
126
+
}
127
+
128
+
// handleQueryLabels implements com.atproto.label.queryLabels (HTTP GET).
129
+
func (s *Server) handleQueryLabels(w http.ResponseWriter, r *http.Request) {
130
+
uriPatterns := r.URL.Query()["uriPatterns"]
131
+
cursorStr := r.URL.Query().Get("cursor")
132
+
limitStr := r.URL.Query().Get("limit")
133
+
134
+
var cursor int64
135
+
if cursorStr != "" {
136
+
cursor, _ = strconv.ParseInt(cursorStr, 10, 64)
137
+
}
138
+
limit := 50
139
+
if limitStr != "" {
140
+
if l, err := strconv.Atoi(limitStr); err == nil && l > 0 && l <= 250 {
141
+
limit = l
142
+
}
143
+
}
144
+
145
+
labels, err := GetLabelsSince(s.db, cursor, limit)
146
+
if err != nil {
147
+
http.Error(w, "failed to query labels", http.StatusInternalServerError)
148
+
return
149
+
}
150
+
151
+
// Filter by URI patterns if provided
152
+
var filtered []LabelOutput
153
+
for _, l := range labels {
154
+
if len(uriPatterns) == 0 || matchesAnyPattern(l.URI, uriPatterns) {
155
+
filtered = append(filtered, labelToOutput(l))
156
+
}
157
+
}
158
+
159
+
var nextCursor string
160
+
if len(labels) > 0 {
161
+
nextCursor = strconv.FormatInt(labels[len(labels)-1].ID, 10)
162
+
}
163
+
164
+
resp := struct {
165
+
Cursor string `json:"cursor,omitempty"`
166
+
Labels []LabelOutput `json:"labels"`
167
+
}{
168
+
Cursor: nextCursor,
169
+
Labels: filtered,
170
+
}
171
+
if resp.Labels == nil {
172
+
resp.Labels = []LabelOutput{}
173
+
}
174
+
175
+
w.Header().Set("Content-Type", "application/json")
176
+
_ = json.NewEncoder(w).Encode(resp)
177
+
}
178
+
179
+
func matchesAnyPattern(uri string, patterns []string) bool {
180
+
for _, p := range patterns {
181
+
// Simple prefix matching (ATProto spec allows glob-like patterns)
182
+
if p == uri || (len(p) > 0 && p[len(p)-1] == '*' && len(uri) >= len(p)-1 && uri[:len(p)-1] == p[:len(p)-1]) {
183
+
return true
184
+
}
185
+
}
186
+
return false
187
+
}
+86
pkg/labeler/subscribe_test.go
+86
pkg/labeler/subscribe_test.go
···
1
+
package labeler
2
+
3
+
import (
4
+
"testing"
5
+
"time"
6
+
)
7
+
8
+
func TestLabelToOutput(t *testing.T) {
9
+
now := time.Date(2026, 3, 22, 10, 0, 0, 0, time.UTC)
10
+
exp := time.Date(2026, 4, 22, 10, 0, 0, 0, time.UTC)
11
+
12
+
label := Label{
13
+
ID: 1,
14
+
Src: "did:web:labeler.atcr.io",
15
+
URI: "at://did:plc:abc/io.atcr.manifest/sha256-123",
16
+
CID: "bafyabc",
17
+
Val: "!takedown",
18
+
Neg: false,
19
+
Cts: now,
20
+
Exp: &exp,
21
+
SubjectDID: "did:plc:abc",
22
+
SubjectRepo: "myimage",
23
+
}
24
+
25
+
out := labelToOutput(label)
26
+
if out.Src != "did:web:labeler.atcr.io" {
27
+
t.Errorf("Src = %q, want did:web:labeler.atcr.io", out.Src)
28
+
}
29
+
if out.URI != "at://did:plc:abc/io.atcr.manifest/sha256-123" {
30
+
t.Errorf("URI = %q", out.URI)
31
+
}
32
+
if out.CID != "bafyabc" {
33
+
t.Errorf("CID = %q, want bafyabc", out.CID)
34
+
}
35
+
if out.Val != "!takedown" {
36
+
t.Errorf("Val = %q", out.Val)
37
+
}
38
+
if out.Neg {
39
+
t.Error("expected Neg=false")
40
+
}
41
+
if out.Cts != "2026-03-22T10:00:00Z" {
42
+
t.Errorf("Cts = %q", out.Cts)
43
+
}
44
+
if out.Exp != "2026-04-22T10:00:00Z" {
45
+
t.Errorf("Exp = %q", out.Exp)
46
+
}
47
+
}
48
+
49
+
func TestLabelToOutput_NoExpiration(t *testing.T) {
50
+
label := Label{
51
+
Src: "did:web:labeler.atcr.io",
52
+
URI: "at://did:plc:abc",
53
+
Val: "!takedown",
54
+
Cts: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC),
55
+
}
56
+
57
+
out := labelToOutput(label)
58
+
if out.Exp != "" {
59
+
t.Errorf("expected empty Exp, got %q", out.Exp)
60
+
}
61
+
}
62
+
63
+
func TestMatchesAnyPattern(t *testing.T) {
64
+
tests := []struct {
65
+
name string
66
+
uri string
67
+
patterns []string
68
+
want bool
69
+
}{
70
+
{"exact match", "at://did:plc:abc/io.atcr.manifest/sha256-123", []string{"at://did:plc:abc/io.atcr.manifest/sha256-123"}, true},
71
+
{"no match", "at://did:plc:abc/io.atcr.manifest/sha256-123", []string{"at://did:plc:def/io.atcr.manifest/sha256-123"}, false},
72
+
{"wildcard match", "at://did:plc:abc/io.atcr.manifest/sha256-123", []string{"at://did:plc:abc/*"}, true},
73
+
{"wildcard no match", "at://did:plc:abc/io.atcr.manifest/sha256-123", []string{"at://did:plc:def/*"}, false},
74
+
{"empty patterns", "at://did:plc:abc/io.atcr.manifest/sha256-123", []string{}, false},
75
+
{"multiple patterns", "at://did:plc:abc/io.atcr.manifest/sha256-123", []string{"at://did:plc:def/*", "at://did:plc:abc/*"}, true},
76
+
}
77
+
78
+
for _, tt := range tests {
79
+
t.Run(tt.name, func(t *testing.T) {
80
+
got := matchesAnyPattern(tt.uri, tt.patterns)
81
+
if got != tt.want {
82
+
t.Errorf("matchesAnyPattern(%q, %v) = %v, want %v", tt.uri, tt.patterns, got, tt.want)
83
+
}
84
+
})
85
+
}
86
+
}
+418
pkg/labeler/takedown.go
+418
pkg/labeler/takedown.go
···
1
+
package labeler
2
+
3
+
import (
4
+
"context"
5
+
"encoding/json"
6
+
"fmt"
7
+
"html/template"
8
+
"log/slog"
9
+
"net/http"
10
+
"strings"
11
+
"time"
12
+
13
+
"atcr.io/pkg/atproto"
14
+
)
15
+
16
+
// TakedownInput represents parsed takedown input.
17
+
type TakedownInput struct {
18
+
DID string
19
+
Handle string
20
+
Repository string // empty = user-level takedown
21
+
}
22
+
23
+
// ParseTakedownInput parses various input formats into a TakedownInput.
24
+
// Supported formats:
25
+
// - atcr.io/r/handle/repo
26
+
// - handle/repo
27
+
// - at://did:plc:xyz/io.atcr.repo.page/repo
28
+
// - at://did:plc:xyz (user-level)
29
+
// - handle (user-level)
30
+
// - did:plc:xyz (user-level)
31
+
func ParseTakedownInput(ctx context.Context, input string) (*TakedownInput, error) {
32
+
input = strings.TrimSpace(input)
33
+
34
+
// AT URI format
35
+
if strings.HasPrefix(input, "at://") {
36
+
return parseATURI(ctx, input)
37
+
}
38
+
39
+
// Strip URL prefix if present
40
+
input = strings.TrimPrefix(input, "https://")
41
+
input = strings.TrimPrefix(input, "http://")
42
+
43
+
// Remove atcr.io/r/ or similar prefix
44
+
for _, prefix := range []string{"atcr.io/r/", "localhost/r/"} {
45
+
if strings.HasPrefix(input, prefix) {
46
+
input = strings.TrimPrefix(input, prefix)
47
+
break
48
+
}
49
+
}
50
+
// Also handle custom domains: anything ending in /r/
51
+
if idx := strings.Index(input, "/r/"); idx >= 0 {
52
+
input = input[idx+3:]
53
+
}
54
+
55
+
// Now input should be "handle/repo" or "handle" or "did:xxx"
56
+
parts := strings.SplitN(input, "/", 2)
57
+
identifier := parts[0]
58
+
var repo string
59
+
if len(parts) > 1 {
60
+
repo = parts[1]
61
+
repo = strings.TrimSuffix(repo, "/")
62
+
}
63
+
64
+
did, handle, err := resolveIdentifier(ctx, identifier)
65
+
if err != nil {
66
+
return nil, err
67
+
}
68
+
69
+
return &TakedownInput{
70
+
DID: did,
71
+
Handle: handle,
72
+
Repository: repo,
73
+
}, nil
74
+
}
75
+
76
+
func parseATURI(ctx context.Context, uri string) (*TakedownInput, error) {
77
+
// at://did:plc:xyz/collection/rkey
78
+
trimmed := strings.TrimPrefix(uri, "at://")
79
+
parts := strings.SplitN(trimmed, "/", 3)
80
+
81
+
did := parts[0]
82
+
if !strings.HasPrefix(did, "did:") {
83
+
// It's a handle
84
+
resolvedDID, handle, err := resolveIdentifier(ctx, did)
85
+
if err != nil {
86
+
return nil, err
87
+
}
88
+
did = resolvedDID
89
+
if len(parts) >= 3 {
90
+
return &TakedownInput{DID: did, Handle: handle, Repository: parts[2]}, nil
91
+
}
92
+
return &TakedownInput{DID: did, Handle: handle}, nil
93
+
}
94
+
95
+
// Resolve handle from DID
96
+
_, handle, _, _ := atproto.ResolveIdentity(ctx, did)
97
+
98
+
if len(parts) < 3 {
99
+
// User-level takedown
100
+
return &TakedownInput{DID: did, Handle: handle}, nil
101
+
}
102
+
103
+
// Extract repository from rkey (third part)
104
+
repo := parts[2]
105
+
return &TakedownInput{DID: did, Handle: handle, Repository: repo}, nil
106
+
}
107
+
108
+
func resolveIdentifier(ctx context.Context, identifier string) (did, handle string, err error) {
109
+
did, handle, _, err = atproto.ResolveIdentity(ctx, identifier)
110
+
if err != nil {
111
+
return "", "", fmt.Errorf("failed to resolve %q: %w", identifier, err)
112
+
}
113
+
return did, handle, nil
114
+
}
115
+
116
+
// TakedownResult contains the results of a takedown operation.
117
+
type TakedownResult struct {
118
+
DID string
119
+
Handle string
120
+
Repository string
121
+
Labels []Label
122
+
UserLevel bool
123
+
}
124
+
125
+
// ExecuteTakedown creates takedown labels for a repo or user.
126
+
func (s *Server) ExecuteTakedown(ctx context.Context, input *TakedownInput) (*TakedownResult, error) {
127
+
src := s.config.DID()
128
+
now := time.Now().UTC()
129
+
result := &TakedownResult{
130
+
DID: input.DID,
131
+
Handle: input.Handle,
132
+
Repository: input.Repository,
133
+
UserLevel: input.Repository == "",
134
+
}
135
+
136
+
if input.Repository == "" {
137
+
// User-level takedown
138
+
label := &Label{
139
+
Src: src,
140
+
URI: "at://" + input.DID,
141
+
Val: "!takedown",
142
+
Cts: now,
143
+
SubjectDID: input.DID,
144
+
SubjectRepo: "",
145
+
}
146
+
if _, err := CreateLabel(s.db, label); err != nil {
147
+
return nil, fmt.Errorf("failed to create user-level label: %w", err)
148
+
}
149
+
result.Labels = append(result.Labels, *label)
150
+
slog.Info("Created user-level takedown", "did", input.DID, "handle", input.Handle)
151
+
return result, nil
152
+
}
153
+
154
+
// Repo-level takedown: discover all records from PDS
155
+
labels, err := s.discoverAndLabelRecords(ctx, input.DID, input.Repository, src, now)
156
+
if err != nil {
157
+
// Even if PDS discovery fails, create a repo-level summary label
158
+
slog.Warn("PDS discovery failed, creating summary label only", "error", err)
159
+
}
160
+
result.Labels = append(result.Labels, labels...)
161
+
162
+
// Always create a repo-level summary label for efficient filtering
163
+
summaryLabel := &Label{
164
+
Src: src,
165
+
URI: fmt.Sprintf("at://%s/io.atcr.repo/%s", input.DID, input.Repository),
166
+
Val: "!takedown",
167
+
Cts: now,
168
+
SubjectDID: input.DID,
169
+
SubjectRepo: input.Repository,
170
+
}
171
+
if _, err := CreateLabel(s.db, summaryLabel); err != nil {
172
+
return nil, fmt.Errorf("failed to create summary label: %w", err)
173
+
}
174
+
result.Labels = append(result.Labels, *summaryLabel)
175
+
176
+
slog.Info("Created repo-level takedown",
177
+
"did", input.DID,
178
+
"handle", input.Handle,
179
+
"repository", input.Repository,
180
+
"label_count", len(result.Labels),
181
+
)
182
+
183
+
return result, nil
184
+
}
185
+
186
+
// discoverAndLabelRecords queries the user's PDS for all records in the given repo
187
+
// and creates takedown labels for each.
188
+
func (s *Server) discoverAndLabelRecords(ctx context.Context, did, repo, src string, now time.Time) ([]Label, error) {
189
+
_, _, pdsEndpoint, err := atproto.ResolveIdentity(ctx, did)
190
+
if err != nil {
191
+
return nil, fmt.Errorf("failed to resolve DID: %w", err)
192
+
}
193
+
194
+
client := atproto.NewClient(pdsEndpoint, did, "")
195
+
var labels []Label
196
+
197
+
// Collections to search
198
+
collections := []string{
199
+
atproto.ManifestCollection,
200
+
atproto.TagCollection,
201
+
atproto.RepoPageCollection,
202
+
}
203
+
204
+
for _, collection := range collections {
205
+
records, _, err := client.ListRecordsForRepo(ctx, did, collection, 100, "")
206
+
if err != nil {
207
+
slog.Warn("Failed to list records", "collection", collection, "error", err)
208
+
continue
209
+
}
210
+
211
+
for _, rec := range records {
212
+
// Filter by repository field
213
+
recRepo := extractRepoField(rec.Value, collection)
214
+
if recRepo != repo {
215
+
continue
216
+
}
217
+
218
+
// Use the full AT URI from the record (at://did/collection/rkey)
219
+
uri := rec.URI
220
+
label := &Label{
221
+
Src: src,
222
+
URI: uri,
223
+
Val: "!takedown",
224
+
Cts: now,
225
+
SubjectDID: did,
226
+
SubjectRepo: repo,
227
+
}
228
+
if _, err := CreateLabel(s.db, label); err != nil {
229
+
slog.Warn("Failed to create label", "uri", uri, "error", err)
230
+
continue
231
+
}
232
+
labels = append(labels, *label)
233
+
}
234
+
}
235
+
236
+
return labels, nil
237
+
}
238
+
239
+
// extractRepoField extracts the repository name from a record's JSON value.
240
+
func extractRepoField(value json.RawMessage, collection string) string {
241
+
// For repo pages, the rkey IS the repository name, but we also check the value
242
+
var rec struct {
243
+
Repository string `json:"repository"`
244
+
}
245
+
if err := json.Unmarshal(value, &rec); err == nil && rec.Repository != "" {
246
+
return rec.Repository
247
+
}
248
+
return ""
249
+
}
250
+
251
+
// Handlers
252
+
253
+
func (s *Server) handleDashboard(w http.ResponseWriter, r *http.Request) {
254
+
labels, total, err := ListActiveTakedowns(s.db, 50, 0)
255
+
if err != nil {
256
+
http.Error(w, "Failed to list takedowns", http.StatusInternalServerError)
257
+
return
258
+
}
259
+
260
+
w.Header().Set("Content-Type", "text/html")
261
+
fmt.Fprintf(w, `<!DOCTYPE html>
262
+
<html>
263
+
<head><title>%s Labeler</title>
264
+
<style>
265
+
body{font-family:system-ui;max-width:900px;margin:40px auto;padding:0 20px}
266
+
table{width:100%%;border-collapse:collapse;margin:20px 0}
267
+
th,td{text-align:left;padding:8px;border-bottom:1px solid #ddd}
268
+
th{background:#f5f5f5}
269
+
.badge{background:#dc2626;color:white;padding:2px 8px;border-radius:4px;font-size:0.85em}
270
+
a{color:#2563eb}
271
+
nav{display:flex;gap:16px;margin-bottom:24px}
272
+
.btn{padding:8px 16px;background:#2563eb;color:white;text-decoration:none;border-radius:4px;border:none;cursor:pointer}
273
+
.btn-danger{background:#dc2626}
274
+
form{display:inline}
275
+
</style>
276
+
</head>
277
+
<body>
278
+
<h1>%s Labeler</h1>
279
+
<nav>
280
+
<a href="/" class="btn">Dashboard</a>
281
+
<a href="/takedown" class="btn">New Takedown</a>
282
+
<a href="/auth/logout">Logout</a>
283
+
</nav>
284
+
<h2>Active Takedowns (%d)</h2>`,
285
+
s.config.Server.ClientShortName,
286
+
s.config.Server.ClientShortName,
287
+
total,
288
+
)
289
+
290
+
if len(labels) == 0 {
291
+
fmt.Fprint(w, `<p>No active takedowns.</p>`)
292
+
} else {
293
+
fmt.Fprint(w, `<table><tr><th>Subject</th><th>Repository</th><th>URI</th><th>Created</th><th>Action</th></tr>`)
294
+
for _, l := range labels {
295
+
repoDisplay := l.SubjectRepo
296
+
if repoDisplay == "" {
297
+
repoDisplay = "<em>all repos (user-level)</em>"
298
+
}
299
+
fmt.Fprintf(w, `<tr>
300
+
<td>%s</td>
301
+
<td>%s</td>
302
+
<td><code>%s</code></td>
303
+
<td>%s</td>
304
+
<td><form method="POST" action="/reverse"><input type="hidden" name="did" value="%s"><input type="hidden" name="repo" value="%s"><button type="submit" class="btn btn-danger" onclick="return confirm('Reverse this takedown?')">Reverse</button></form></td>
305
+
</tr>`,
306
+
template.HTMLEscapeString(l.SubjectDID),
307
+
repoDisplay,
308
+
template.HTMLEscapeString(l.URI),
309
+
l.Cts.Format("2006-01-02 15:04"),
310
+
template.HTMLEscapeString(l.SubjectDID),
311
+
template.HTMLEscapeString(l.SubjectRepo),
312
+
)
313
+
}
314
+
fmt.Fprint(w, `</table>`)
315
+
}
316
+
317
+
fmt.Fprint(w, `</body></html>`)
318
+
}
319
+
320
+
func (s *Server) handleTakedownForm(w http.ResponseWriter, r *http.Request) {
321
+
msg := r.URL.Query().Get("msg")
322
+
errorMsg := r.URL.Query().Get("error")
323
+
324
+
w.Header().Set("Content-Type", "text/html")
325
+
fmt.Fprintf(w, `<!DOCTYPE html>
326
+
<html>
327
+
<head><title>%s Labeler - New Takedown</title>
328
+
<style>
329
+
body{font-family:system-ui;max-width:600px;margin:40px auto;padding:0 20px}
330
+
input[type=text]{width:100%%;padding:8px;margin:8px 0;box-sizing:border-box}
331
+
.btn{padding:10px 20px;background:#dc2626;color:white;border:none;border-radius:4px;cursor:pointer;font-size:1em}
332
+
.success{color:green;margin-bottom:1em}
333
+
.error{color:red;margin-bottom:1em}
334
+
a{color:#2563eb}
335
+
nav{display:flex;gap:16px;margin-bottom:24px}
336
+
.nav-btn{padding:8px 16px;background:#2563eb;color:white;text-decoration:none;border-radius:4px}
337
+
.help{color:#666;font-size:0.9em;margin-top:4px}
338
+
</style>
339
+
</head>
340
+
<body>
341
+
<h1>New Takedown</h1>
342
+
<nav>
343
+
<a href="/" class="nav-btn">Dashboard</a>
344
+
<a href="/takedown" class="nav-btn">New Takedown</a>
345
+
</nav>`,
346
+
s.config.Server.ClientShortName,
347
+
)
348
+
349
+
if msg != "" {
350
+
fmt.Fprintf(w, `<div class="success">%s</div>`, template.HTMLEscapeString(msg))
351
+
}
352
+
if errorMsg != "" {
353
+
fmt.Fprintf(w, `<div class="error">%s</div>`, template.HTMLEscapeString(errorMsg))
354
+
}
355
+
356
+
fmt.Fprint(w, `
357
+
<form method="POST" action="/takedown">
358
+
<label for="target"><strong>Target</strong></label>
359
+
<input type="text" id="target" name="target" placeholder="atcr.io/r/handle/repo, at://did/collection/rkey, or handle" required>
360
+
<p class="help">Accepts repo URLs, AT URIs, handles, or DIDs. Omit the repo for a user-level takedown.</p>
361
+
<br>
362
+
<button type="submit" class="btn" onclick="return confirm('Issue takedown? This will suppress the content immediately.')">Issue Takedown</button>
363
+
</form>
364
+
</body></html>`)
365
+
}
366
+
367
+
func (s *Server) handleTakedownSubmit(w http.ResponseWriter, r *http.Request) {
368
+
target := strings.TrimSpace(r.FormValue("target"))
369
+
if target == "" {
370
+
http.Redirect(w, r, "/takedown?error=Target+is+required", http.StatusFound)
371
+
return
372
+
}
373
+
374
+
input, err := ParseTakedownInput(r.Context(), target)
375
+
if err != nil {
376
+
http.Redirect(w, r, "/takedown?error="+strings.ReplaceAll(err.Error(), " ", "+"), http.StatusFound)
377
+
return
378
+
}
379
+
380
+
result, err := s.ExecuteTakedown(r.Context(), input)
381
+
if err != nil {
382
+
http.Redirect(w, r, "/takedown?error="+strings.ReplaceAll(err.Error(), " ", "+"), http.StatusFound)
383
+
return
384
+
}
385
+
386
+
msg := fmt.Sprintf("Takedown issued: %d labels created for %s", len(result.Labels), result.DID)
387
+
if result.Repository != "" {
388
+
msg += "/" + result.Repository
389
+
}
390
+
http.Redirect(w, r, "/takedown?msg="+strings.ReplaceAll(msg, " ", "+"), http.StatusFound)
391
+
}
392
+
393
+
func (s *Server) handleReverse(w http.ResponseWriter, r *http.Request) {
394
+
did := strings.TrimSpace(r.FormValue("did"))
395
+
repo := strings.TrimSpace(r.FormValue("repo"))
396
+
397
+
if did == "" {
398
+
http.Redirect(w, r, "/?error=DID+is+required", http.StatusFound)
399
+
return
400
+
}
401
+
402
+
src := s.config.DID()
403
+
var err error
404
+
if repo == "" {
405
+
err = NegateUserLabels(s.db, src, did)
406
+
} else {
407
+
err = NegateRepoLabels(s.db, src, did, repo)
408
+
}
409
+
410
+
if err != nil {
411
+
slog.Error("Failed to reverse takedown", "did", did, "repo", repo, "error", err)
412
+
http.Redirect(w, r, "/?error=Failed+to+reverse+takedown", http.StatusFound)
413
+
return
414
+
}
415
+
416
+
slog.Info("Reversed takedown", "did", did, "repo", repo)
417
+
http.Redirect(w, r, "/", http.StatusFound)
418
+
}
+139
pkg/labeler/takedown_test.go
+139
pkg/labeler/takedown_test.go
···
1
+
package labeler
2
+
3
+
import (
4
+
"context"
5
+
"testing"
6
+
)
7
+
8
+
func TestParseTakedownInput_RepoURL(t *testing.T) {
9
+
// These tests only exercise parsing logic, not PDS resolution.
10
+
// ResolveIdentity calls are tested with mock server below.
11
+
tests := []struct {
12
+
name string
13
+
input string
14
+
wantRepo string
15
+
}{
16
+
{"full URL", "https://atcr.io/r/handle/myimage", "myimage"},
17
+
{"no scheme", "atcr.io/r/handle/myimage", "myimage"},
18
+
{"handle/repo", "handle/myimage", "myimage"},
19
+
{"trailing slash", "atcr.io/r/handle/myimage/", "myimage"},
20
+
{"custom domain", "https://registry.example.com/r/handle/myimage", "myimage"},
21
+
}
22
+
23
+
for _, tt := range tests {
24
+
t.Run(tt.name, func(t *testing.T) {
25
+
// These will fail on ResolveIdentity since there's no real PDS,
26
+
// but we can at least verify the parsing doesn't panic
27
+
_, err := ParseTakedownInput(context.Background(), tt.input)
28
+
if err == nil {
29
+
t.Skip("ResolveIdentity succeeded unexpectedly (network available)")
30
+
}
31
+
// The error should be from resolution, not parsing
32
+
if err != nil {
33
+
t.Logf("Expected resolution error: %v", err)
34
+
}
35
+
})
36
+
}
37
+
}
38
+
39
+
func TestParseTakedownInput_ATURI(t *testing.T) {
40
+
tests := []struct {
41
+
name string
42
+
input string
43
+
wantDID string
44
+
wantRepo string
45
+
}{
46
+
{
47
+
"full AT URI with collection",
48
+
"at://did:plc:abc123/io.atcr.repo.page/myimage",
49
+
"did:plc:abc123",
50
+
"myimage",
51
+
},
52
+
{
53
+
"DID only (user-level)",
54
+
"at://did:plc:abc123",
55
+
"did:plc:abc123",
56
+
"",
57
+
},
58
+
{
59
+
"DID with collection and rkey",
60
+
"at://did:plc:xyz/io.atcr.manifest/sha256-deadbeef",
61
+
"did:plc:xyz",
62
+
"sha256-deadbeef",
63
+
},
64
+
}
65
+
66
+
for _, tt := range tests {
67
+
t.Run(tt.name, func(t *testing.T) {
68
+
input, err := ParseTakedownInput(context.Background(), tt.input)
69
+
if err != nil {
70
+
// Resolution may fail for handle-based AT URIs
71
+
t.Logf("Parse error (may be expected): %v", err)
72
+
return
73
+
}
74
+
if input.DID != tt.wantDID {
75
+
t.Errorf("DID = %q, want %q", input.DID, tt.wantDID)
76
+
}
77
+
if input.Repository != tt.wantRepo {
78
+
t.Errorf("Repository = %q, want %q", input.Repository, tt.wantRepo)
79
+
}
80
+
})
81
+
}
82
+
}
83
+
84
+
func TestParseTakedownInput_DID(t *testing.T) {
85
+
// Direct DID input (user-level takedown)
86
+
input, err := ParseTakedownInput(context.Background(), "at://did:plc:abc123")
87
+
if err != nil {
88
+
t.Fatalf("unexpected error: %v", err)
89
+
}
90
+
if input.DID != "did:plc:abc123" {
91
+
t.Errorf("DID = %q, want did:plc:abc123", input.DID)
92
+
}
93
+
if input.Repository != "" {
94
+
t.Errorf("Repository = %q, want empty (user-level)", input.Repository)
95
+
}
96
+
}
97
+
98
+
func TestExtractRepoField(t *testing.T) {
99
+
tests := []struct {
100
+
name string
101
+
value string
102
+
collection string
103
+
want string
104
+
}{
105
+
{
106
+
"manifest with repository",
107
+
`{"$type":"io.atcr.manifest","repository":"myimage","digest":"sha256:abc"}`,
108
+
"io.atcr.manifest",
109
+
"myimage",
110
+
},
111
+
{
112
+
"tag with repository",
113
+
`{"$type":"io.atcr.tag","repository":"myimage","tag":"latest"}`,
114
+
"io.atcr.tag",
115
+
"myimage",
116
+
},
117
+
{
118
+
"no repository field",
119
+
`{"$type":"io.atcr.manifest","digest":"sha256:abc"}`,
120
+
"io.atcr.manifest",
121
+
"",
122
+
},
123
+
{
124
+
"invalid JSON",
125
+
`{invalid}`,
126
+
"io.atcr.manifest",
127
+
"",
128
+
},
129
+
}
130
+
131
+
for _, tt := range tests {
132
+
t.Run(tt.name, func(t *testing.T) {
133
+
got := extractRepoField([]byte(tt.value), tt.collection)
134
+
if got != tt.want {
135
+
t.Errorf("extractRepoField() = %q, want %q", got, tt.want)
136
+
}
137
+
})
138
+
}
139
+
}