this repo has no description
Merge
+81
-3265
-54
.github/workflows/container-labelmaker-ghcr.yaml
-54
.github/workflows/container-labelmaker-ghcr.yaml
···
1
-
name: container-labelmaker-ghcr
2
-
on:
3
-
push:
4
-
branches:
5
-
- main
6
-
- bnewbold/labelmaker-more
7
-
env:
8
-
REGISTRY: ghcr.io
9
-
# github.repository as <account>/<repo>
10
-
IMAGE_NAME: ${{ github.repository }}
11
-
12
-
jobs:
13
-
container-labelmaker-ghcr:
14
-
if: github.repository == 'bluesky-social/indigo'
15
-
runs-on: ubuntu-latest
16
-
permissions:
17
-
contents: read
18
-
packages: write
19
-
id-token: write
20
-
21
-
steps:
22
-
- name: Checkout repository
23
-
uses: actions/checkout@v3
24
-
25
-
- name: Setup Docker buildx
26
-
uses: docker/setup-buildx-action@v1
27
-
28
-
- name: Log into registry ${{ env.REGISTRY }}
29
-
uses: docker/login-action@v2
30
-
with:
31
-
registry: ${{ env.REGISTRY }}
32
-
username: ${{ github.actor }}
33
-
password: ${{ secrets.GITHUB_TOKEN }}
34
-
35
-
- name: Extract Docker metadata
36
-
id: meta
37
-
uses: docker/metadata-action@v4
38
-
with:
39
-
images: |
40
-
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
41
-
tags: |
42
-
type=sha,enable=true,priority=100,prefix=labelmaker:,suffix=,format=long
43
-
44
-
- name: Build and push Docker image
45
-
id: build-and-push
46
-
uses: docker/build-push-action@v4
47
-
with:
48
-
context: .
49
-
file: ./cmd/labelmaker/Dockerfile
50
-
push: ${{ github.event_name != 'pull_request' }}
51
-
tags: ${{ steps.meta.outputs.tags }}
52
-
labels: ${{ steps.meta.outputs.labels }}
53
-
cache-from: type=gha
54
-
cache-to: type=gha,mode=max
-1
.gitignore
-1
.gitignore
+2
-3
HACKING.md
+2
-3
HACKING.md
···
114
114
115
115
## Integrated Development
116
116
117
-
Sometimes it is helpful to run a PLC, PDS, BGS, labelmaker, and other components, all locally on your laptop, across languages. This section describes one setup for this.
117
+
Sometimes it is helpful to run a PLC, PDS, BGS, and other components, all locally on your laptop, across languages. This section describes one setup for this.
118
118
119
119
First, you need PostgreSQL running locally. This could be via docker, or the following commands assume some kind of debian/ubuntu setup with a postgres server package installed and running.
120
120
···
139
139
140
140
make run-dev-pds
141
141
142
-
In this repo (indigo), start a BGS and labelmaker, in two separate terminals:
142
+
In this repo (indigo), start a BGS, in two separate terminals:
143
143
144
144
make run-dev-bgs
145
-
make run-dev-labelmaker
146
145
147
146
In a final terminal, run fakermaker to inject data into the system:
148
147
-5
Makefile
-5
Makefile
···
23
23
go build ./cmd/lexgen
24
24
go build ./cmd/stress
25
25
go build ./cmd/fakermaker
26
-
go build ./cmd/labelmaker
27
26
go build ./cmd/hepa
28
27
go build ./cmd/supercollider
29
28
go build -o ./sonar-cli ./cmd/sonar
···
91
90
run-bgs-image:
92
91
docker run -p 2470:2470 bigsky /bigsky --admin-key localdev
93
92
# --crawl-insecure-ws
94
-
95
-
.PHONY: run-dev-labelmaker
96
-
run-dev-labelmaker: .env ## Runs labelmaker for local dev
97
-
GOLOG_LOG_LEVEL=info go run ./cmd/labelmaker --subscribe-insecure-ws
98
93
99
94
.PHONY: run-dev-search
100
95
run-dev-search: .env ## Runs search daemon for local dev
+2
-2
README.md
+2
-2
README.md
···
53
53
54
54
*not to be confused with the [AT command set](https://en.wikipedia.org/wiki/Hayes_command_set) or [Adenosine triphosphate](https://en.wikipedia.org/wiki/Adenosine_triphosphate)*
55
55
56
-
The Authenticated Transfer Protocol ("ATP" or "atproto") is a decentralized social media protocol, developed by [Bluesky PBC](https://blueskyweb.xyz). Learn more at:
56
+
The Authenticated Transfer Protocol ("ATP" or "atproto") is a decentralized social media protocol, developed by [Bluesky PBC](https://bsky.social). Learn more at:
57
57
58
58
- [Overview and Guides](https://atproto.com/guides/overview) 👈 Best starting point
59
59
- [Github Discussions](https://github.com/bluesky-social/atproto/discussions) 👈 Great place to ask questions
60
60
- [Protocol Specifications](https://atproto.com/specs/atp)
61
-
- [Blogpost on self-authenticating data structures](https://blueskyweb.xyz/blog/3-6-2022-a-self-authenticating-social-protocol)
61
+
- [Blogpost on self-authenticating data structures](https://bsky.social/about/blog/3-6-2022-a-self-authenticating-social-protocol)
62
62
63
63
The Bluesky Social application encompasses a set of schemas and APIs built in the overall AT Protocol framework. The namespace for these "Lexicons" is `app.bsky.*`.
64
64
+10
-5
atproto/identity/cache_directory.go
+10
-5
atproto/identity/cache_directory.go
···
16
16
type CacheDirectory struct {
17
17
Inner Directory
18
18
ErrTTL time.Duration
19
+
InvalidHandleTTL time.Duration
19
20
handleCache *expirable.LRU[syntax.Handle, HandleEntry]
20
21
identityCache *expirable.LRU[syntax.DID, IdentityEntry]
21
22
didLookupChans sync.Map
···
67
68
var _ Directory = (*CacheDirectory)(nil)
68
69
69
70
// Capacity of zero means unlimited size. Similarly, ttl of zero means unlimited duration.
70
-
func NewCacheDirectory(inner Directory, capacity int, hitTTL, errTTL time.Duration) CacheDirectory {
71
+
func NewCacheDirectory(inner Directory, capacity int, hitTTL, errTTL time.Duration, invalidHandleTTL time.Duration) CacheDirectory {
71
72
return CacheDirectory{
72
-
ErrTTL: errTTL,
73
-
Inner: inner,
74
-
handleCache: expirable.NewLRU[syntax.Handle, HandleEntry](capacity, nil, hitTTL),
75
-
identityCache: expirable.NewLRU[syntax.DID, IdentityEntry](capacity, nil, hitTTL),
73
+
ErrTTL: errTTL,
74
+
InvalidHandleTTL: invalidHandleTTL,
75
+
Inner: inner,
76
+
handleCache: expirable.NewLRU[syntax.Handle, HandleEntry](capacity, nil, hitTTL),
77
+
identityCache: expirable.NewLRU[syntax.DID, IdentityEntry](capacity, nil, hitTTL),
76
78
}
77
79
}
78
80
···
85
87
86
88
func (d *CacheDirectory) IsIdentityStale(e *IdentityEntry) bool {
87
89
if e.Err != nil && time.Since(e.Updated) > d.ErrTTL {
90
+
return true
91
+
}
92
+
if e.Identity != nil && e.Identity.Handle.IsInvalidHandle() && time.Since(e.Updated) > d.InvalidHandleTTL {
88
93
return true
89
94
}
90
95
return false
+1
-1
atproto/identity/identity.go
+1
-1
atproto/identity/identity.go
···
76
76
// primary Bluesky PDS instance only supports HTTP resolution method
77
77
SkipDNSDomainSuffixes: []string{".bsky.social"},
78
78
}
79
-
cached := NewCacheDirectory(&base, 10000, time.Hour*24, time.Minute*2)
79
+
cached := NewCacheDirectory(&base, 250_000, time.Hour*24, time.Minute*2, time.Minute*5)
80
80
return &cached
81
81
}
82
82
+2
-2
atproto/identity/live_test.go
+2
-2
atproto/identity/live_test.go
···
64
64
func TestCacheDirectory(t *testing.T) {
65
65
t.Skip("TODO: skipping live network test")
66
66
inner := BaseDirectory{}
67
-
d := NewCacheDirectory(&inner, 1000, time.Hour*1, time.Hour*1)
67
+
d := NewCacheDirectory(&inner, 1000, time.Hour*1, time.Hour*1, time.Hour*1)
68
68
for i := 0; i < 3; i = i + 1 {
69
69
testDirectoryLive(t, &d)
70
70
}
···
87
87
TryAuthoritativeDNS: true,
88
88
SkipDNSDomainSuffixes: []string{".bsky.social"},
89
89
}
90
-
dir := NewCacheDirectory(&base, 1000, time.Hour*1, time.Hour*1)
90
+
dir := NewCacheDirectory(&base, 1000, time.Hour*1, time.Hour*1, time.Hour*1)
91
91
// All 60 routines launch at the same time, so they should all miss the cache initially
92
92
routines := 60
93
93
wg := sync.WaitGroup{}
+1
-1
backfill/gormstore.go
+1
-1
backfill/gormstore.go
+1
-1
cmd/hepa/README.md
+1
-1
cmd/hepa/README.md
···
15
15
- which rules are included configured at compile time
16
16
- admin access to fetch private account metadata, and to persist moderation actions, is optional. it is possible for anybody to run a `hepa` instance
17
17
18
-
This is not a "labeling service" per say, in that it pushes labels in to an existing moderation service, and doesn't provide API endpoints or label streams. see `labelmaker` for a self-contained labeling service.
18
+
This is not a "labeling service" per say, in that it pushes labels in to an existing moderation service, and doesn't provide API endpoints or label streams.
19
19
20
20
Performance is generally slow when first starting up, because account-level metadata is being fetched (and cached) for every firehose event. After the caches have "warmed up", events are processed faster.
21
21
+1
cmd/hepa/consumer.go
+1
cmd/hepa/consumer.go
···
82
82
// start at higher parallelism (somewhat arbitrary)
83
83
scaleSettings := autoscaling.DefaultAutoscaleSettings()
84
84
scaleSettings.Concurrency = 6
85
+
scaleSettings.MaxConcurrency = 240
85
86
return events.HandleRepoStream(
86
87
ctx, con, autoscaling.NewScheduler(
87
88
scaleSettings,
+1
-1
cmd/hepa/main.go
+1
-1
cmd/hepa/main.go
···
157
157
}
158
158
dir = rdir
159
159
} else {
160
-
cdir := identity.NewCacheDirectory(&baseDir, 1_500_000, time.Hour*24, time.Minute*2)
160
+
cdir := identity.NewCacheDirectory(&baseDir, 1_500_000, time.Hour*24, time.Minute*2, time.Minute*5)
161
161
dir = &cdir
162
162
}
163
163
return dir, nil
-37
cmd/labelmaker/Dockerfile
-37
cmd/labelmaker/Dockerfile
···
1
-
# Run this dockerfile from the top level of the indigo git repository like:
2
-
#
3
-
# podman build -f ./cmd/labelmaker/Dockerfile -t labelmaker .
4
-
5
-
### Compile stage
6
-
FROM golang:1.21-alpine3.18 AS build-env
7
-
RUN apk add --no-cache build-base make git
8
-
9
-
ADD . /dockerbuild
10
-
WORKDIR /dockerbuild
11
-
12
-
# timezone data for alpine builds
13
-
ENV GOEXPERIMENT=loopvar
14
-
RUN GIT_VERSION=$(git describe --tags --long --always) && \
15
-
go build -tags timetzdata -o /labelmaker ./cmd/labelmaker
16
-
17
-
### Run stage
18
-
FROM alpine:3.18
19
-
20
-
RUN apk add --no-cache --update dumb-init ca-certificates
21
-
ENTRYPOINT ["dumb-init", "--"]
22
-
23
-
WORKDIR /
24
-
RUN mkdir -p data/labelmaker
25
-
COPY --from=build-env /labelmaker /
26
-
27
-
# small things to make golang binaries work well under alpine
28
-
ENV GODEBUG=netdns=go
29
-
ENV TZ=Etc/UTC
30
-
31
-
EXPOSE 2210
32
-
33
-
CMD ["/labelmaker"]
34
-
35
-
LABEL org.opencontainers.image.source=https://github.com/bluesky-social/indigo
36
-
LABEL org.opencontainers.image.description="ATP Labeling Service (labelmaker)"
37
-
LABEL org.opencontainers.image.licenses=MIT
-101
cmd/labelmaker/README.md
-101
cmd/labelmaker/README.md
···
1
-
2
-
labelmaker
3
-
===========
4
-
5
-
## Database Setup
6
-
7
-
PostgreSQL and Sqlite are both supported. When using Sqlite, separate database
8
-
for the BGS database itself and the CarStore are used. With PostgreSQL a single
9
-
database server, user, and database, can all be reused.
10
-
11
-
Database configuration is passed via the `DATABASE_URL` and
12
-
`CARSTORE_DATABASE_URL` environment variables, or the corresponding CLI args.
13
-
14
-
For PostgreSQL, the user and database must already be configured. Some example
15
-
SQL commands are:
16
-
17
-
CREATE DATABASE bgs;
18
-
CREATE DATABASE carstore;
19
-
20
-
CREATE USER ${username} WITH PASSWORD '${password}';
21
-
GRANT ALL PRIVILEGES ON DATABASE bgs TO ${username};
22
-
GRANT ALL PRIVILEGES ON DATABASE carstore TO ${username};
23
-
24
-
This service currently uses `gorm` to automatically run database migrations as
25
-
the regular user. There is no concept of running a separate set of migrations
26
-
under more privileged database user.
27
-
28
-
For database performance with many labels, it is important that `LC_COLLATE=C`.
29
-
That is, the string sort behavior must be by byte order.
30
-
31
-
## Keyword Labeler
32
-
33
-
A trivial keyword filter labeler is included. To configure it, create a JSON
34
-
with the same structure as the `example_keywords.json` file in this directory,
35
-
and provide the path to the `--keyword-file` CLI arg (or the corresponding env
36
-
var).
37
-
38
-
The structure is a list of label values ("value"), each with a list of
39
-
lower-case keyword tokens. If a token is found in post or profile text, the
40
-
corresponding label is generated.
41
-
42
-
43
-
## micro-NSFW-img Integration
44
-
45
-
`micro_nsfw_img` is a simple image classification tool, useful for integration
46
-
testing and local development. You can HTTP POST and image to it and get a set
47
-
of floating point scores back about whether it is hentai, porn, etc. See more
48
-
at <https://gitlab.com/bnewbold/micro-nsfw-img>.
49
-
50
-
To get it working with labelmaker, download the huge (3+ GByte) dockerfile and
51
-
run it locally:
52
-
53
-
docker pull bnewbold/micro-nsfw-img:latest
54
-
docker run --network host bnewbold/micro-nsfw-img
55
-
56
-
Then configure labelmaker with:
57
-
58
-
# or the '--micro-nsfw-img-url' CLI flag
59
-
LABELMAKER_MICRO_NSFW_IMG_URL="http://localhost:5000/classify-image"
60
-
61
-
62
-
## SQRL Integration
63
-
64
-
SQRL is a moderation system built around a declarative rule language,
65
-
application events, and cached counter values. It is the open source release of
66
-
Smyt, a moderation system acquired and used by Twitter many years ago. See the
67
-
SQRL docs for more: <https://sqrl-lang.github.io/sqrl/index.html>
68
-
69
-
A local SQRL moderation server can be queried by providing `--sqrl-url` (or the
70
-
corresponding env var). Post and Profile records will be passed, wrapped in a
71
-
top-level JSON field `EventData`.
72
-
73
-
An example SQRL ruleset for posts and profiles is provided in `sqrl_example`.
74
-
To use this, checkout the SQRL codebase and get it running, then copy the
75
-
`bsky` folder to the top directory and run:
76
-
77
-
./sqrl serve bsky/main.sqrl
78
-
79
-
Counter state will not persist across restarts unless Redis is configured as
80
-
well.
81
-
82
-
83
-
## Repo Account Setup
84
-
85
-
You'll need a DID and handle for the labelmaker service itself.
86
-
87
-
Generate the secret keys (as JSON files), along with did:key representations,
88
-
and store these in a password manager:
89
-
90
-
go run ./cmd/laputa/ gen-key -o labelmaker_signing.key
91
-
go run ./cmd/gosky/ did didKey --keypath labeler_signing.key
92
-
93
-
go run ./cmd/laputa/ gen-key -o labelmaker_recovery.key
94
-
go run ./cmd/gosky/ did didKey --keypath labeler_recovery.key
95
-
96
-
Use the result to generate a new DID:
97
-
98
-
go run ./cmd/gosky/ did create --recoverydid did:key:FROMABOVE --signingkey labeler_signing.key your.handle.tld https://your.pds.host
99
-
100
-
The signing key JSON, along with repo handle and DID, can be passed to
101
-
labelmaker via an environment variables.
-5
cmd/labelmaker/example_keywords.json
-5
cmd/labelmaker/example_keywords.json
-271
cmd/labelmaker/main.go
-271
cmd/labelmaker/main.go
···
1
-
package main
2
-
3
-
import (
4
-
"context"
5
-
"os"
6
-
"path/filepath"
7
-
8
-
"github.com/bluesky-social/indigo/carstore"
9
-
"github.com/bluesky-social/indigo/labeler"
10
-
"github.com/bluesky-social/indigo/util/cliutil"
11
-
"github.com/urfave/cli/v2"
12
-
13
-
_ "github.com/joho/godotenv/autoload"
14
-
_ "go.uber.org/automaxprocs"
15
-
16
-
"github.com/carlmjohnson/versioninfo"
17
-
logging "github.com/ipfs/go-log"
18
-
"github.com/whyrusleeping/go-did"
19
-
"gorm.io/plugin/opentelemetry/tracing"
20
-
)
21
-
22
-
var log = logging.Logger("labelmaker")
23
-
24
-
func main() {
25
-
if err := run(os.Args); err != nil {
26
-
log.Fatal(err)
27
-
}
28
-
}
29
-
30
-
func run(args []string) error {
31
-
32
-
app := cli.App{
33
-
Name: "labelmaker",
34
-
Usage: "atproto content labeling daemon",
35
-
Version: versioninfo.Short(),
36
-
}
37
-
38
-
app.Flags = []cli.Flag{
39
-
&cli.StringFlag{
40
-
Name: "db-url",
41
-
Usage: "database connection string for labelmaker database",
42
-
Value: "sqlite://./data/labelmaker/labelmaker.sqlite",
43
-
EnvVars: []string{"DATABASE_URL"},
44
-
},
45
-
&cli.StringFlag{
46
-
Name: "carstore-db-url",
47
-
Usage: "database connection string for carstore database",
48
-
Value: "sqlite://./data/labelmaker/carstore.sqlite",
49
-
EnvVars: []string{"CARSTORE_DATABASE_URL"},
50
-
},
51
-
&cli.BoolFlag{
52
-
Name: "db-tracing",
53
-
},
54
-
&cli.StringFlag{
55
-
Name: "data-dir",
56
-
Usage: "path of directory for CAR files and other data",
57
-
Value: "data/labelmaker",
58
-
EnvVars: []string{"DATA_DIR"},
59
-
},
60
-
&cli.StringFlag{
61
-
Name: "bgs-host",
62
-
Usage: "hostname and port of BGS to subscribe to",
63
-
Value: "localhost:2470",
64
-
EnvVars: []string{"ATP_BGS_HOST"},
65
-
},
66
-
&cli.StringFlag{
67
-
Name: "plc-host",
68
-
Usage: "method, hostname, and port of PLC registry",
69
-
Value: "https://plc.directory",
70
-
EnvVars: []string{"ATP_PLC_HOST"},
71
-
},
72
-
// TODO(bnewbold): this is a temporary hack to fetch our own blobs
73
-
&cli.StringFlag{
74
-
Name: "pds-host",
75
-
Usage: "method, hostname, and port of PDS instance",
76
-
Value: "http://localhost:4849",
77
-
EnvVars: []string{"ATP_PDS_HOST"},
78
-
},
79
-
&cli.BoolFlag{
80
-
Name: "subscribe-insecure-ws",
81
-
Usage: "when connecting to BGS instance, use ws:// instead of wss://",
82
-
},
83
-
&cli.StringFlag{
84
-
Name: "repo-did",
85
-
Usage: "DID for labelmaker repo",
86
-
Value: "did:plc:FAKE",
87
-
EnvVars: []string{"LABELMAKER_REPO_DID"},
88
-
},
89
-
&cli.StringFlag{
90
-
Name: "repo-handle",
91
-
Usage: "handle for labelmaker repo",
92
-
Value: "labelmaker.test",
93
-
EnvVars: []string{"LABELMAKER_REPO_HANDLE"},
94
-
},
95
-
&cli.StringFlag{
96
-
Name: "repo-password",
97
-
Usage: "labelmaker repo password, used as admin password",
98
-
Value: "admin",
99
-
EnvVars: []string{"LABELMAKER_REPO_PASSWORD"},
100
-
},
101
-
&cli.StringFlag{
102
-
Name: "signing-secret-key-jwk",
103
-
Usage: "signing key for labelmaker repo, in JWK serialization",
104
-
EnvVars: []string{"LABELMAKER_SIGNING_SECRET_KEY_JWK"},
105
-
},
106
-
&cli.StringFlag{
107
-
Name: "bind",
108
-
Usage: "IP or address, and port, to listen on for HTTP and WebSocket APIs",
109
-
Value: ":2210",
110
-
EnvVars: []string{"LABELMAKER_BIND"},
111
-
},
112
-
&cli.StringFlag{
113
-
Name: "xrpc-proxy-url",
114
-
Usage: "backend URL to proxy (some) XRPC requests to",
115
-
Value: "http://localhost:2583",
116
-
EnvVars: []string{"ATP_XRPC_PROXY_URL"},
117
-
},
118
-
&cli.StringFlag{
119
-
Name: "xrpc-proxy-admin-password",
120
-
Usage: "admin auth password for XRPC proxy requests",
121
-
Value: "admin",
122
-
EnvVars: []string{"ATP_XRPC_PROXY_ADMIN_PASSWORD"},
123
-
},
124
-
&cli.StringFlag{
125
-
Name: "keyword-file",
126
-
Usage: "keyword filter config, as JSON file",
127
-
EnvVars: []string{"LABELMAKER_KEYWORD_FILE"},
128
-
},
129
-
&cli.StringFlag{
130
-
Name: "micro-nsfw-img-url",
131
-
Usage: "'micro-nsfw-img' classifier endpoint (full URL)",
132
-
EnvVars: []string{"LABELMAKER_MICRO_NSFW_IMG_URL"},
133
-
},
134
-
&cli.StringFlag{
135
-
Name: "hiveai-api-token",
136
-
Usage: "thehive.ai API token",
137
-
EnvVars: []string{"LABELMAKER_HIVEAI_API_TOKEN"},
138
-
},
139
-
&cli.StringFlag{
140
-
Name: "sqrl-url",
141
-
Usage: "SQRL API endpoint (full URL)",
142
-
EnvVars: []string{"LABELMAKER_SQRL_URL"},
143
-
},
144
-
&cli.IntFlag{
145
-
Name: "max-carstore-connections",
146
-
EnvVars: []string{"MAX_CARSTORE_CONNECTIONS"},
147
-
Value: 40,
148
-
},
149
-
&cli.IntFlag{
150
-
Name: "max-metadb-connections",
151
-
EnvVars: []string{"MAX_METADB_CONNECTIONS"},
152
-
Value: 40,
153
-
},
154
-
}
155
-
156
-
app.Action = func(cctx *cli.Context) error {
157
-
158
-
// ensure data directory exists; won't error if it does
159
-
datadir := cctx.String("data-dir")
160
-
csdir := filepath.Join(datadir, "carstore")
161
-
os.MkdirAll(datadir, os.ModePerm)
162
-
repoKeyPath := filepath.Join(datadir, "labelmaker.key")
163
-
164
-
dburl := cctx.String("db-url")
165
-
db, err := cliutil.SetupDatabase(dburl, cctx.Int("max-metadb-connections"))
166
-
if err != nil {
167
-
return err
168
-
}
169
-
170
-
csdburl := cctx.String("carstore-db-url")
171
-
csdb, err := cliutil.SetupDatabase(csdburl, cctx.Int("max-carstore-connections"))
172
-
if err != nil {
173
-
return err
174
-
}
175
-
176
-
if cctx.Bool("db-tracing") {
177
-
if err := db.Use(tracing.NewPlugin()); err != nil {
178
-
return err
179
-
}
180
-
if err := csdb.Use(tracing.NewPlugin()); err != nil {
181
-
return err
182
-
}
183
-
}
184
-
185
-
os.MkdirAll(filepath.Dir(csdir), os.ModePerm)
186
-
cstore, err := carstore.NewCarStore(csdb, csdir)
187
-
if err != nil {
188
-
return err
189
-
}
190
-
191
-
kwlFile := cctx.String("keyword-file")
192
-
var kwl []labeler.KeywordLabeler
193
-
if kwlFile != "" {
194
-
kwl, err = labeler.LoadKeywordFile(kwlFile)
195
-
if err != nil {
196
-
return err
197
-
}
198
-
} else {
199
-
// trivial examples
200
-
kwl = append(kwl, labeler.KeywordLabeler{Value: "meta", Keywords: []string{"bluesky", "atproto"}})
201
-
kwl = append(kwl, labeler.KeywordLabeler{Value: "wordle", Keywords: []string{"wordle"}})
202
-
kwl = append(kwl, labeler.KeywordLabeler{Value: "definite-article", Keywords: []string{"the"}})
203
-
}
204
-
205
-
bgsURL := cctx.String("bgs-host")
206
-
plcURL := cctx.String("plc-host")
207
-
blobPdsURL := cctx.String("pds-host")
208
-
useWss := !cctx.Bool("subscribe-insecure-ws")
209
-
repoDid := cctx.String("repo-did")
210
-
repoHandle := cctx.String("repo-handle")
211
-
repoPassword := cctx.String("repo-password")
212
-
signingSecretKeyJwk := cctx.String("signing-secret-key-jwk")
213
-
bind := cctx.String("bind")
214
-
xrpcProxyURL := cctx.String("xrpc-proxy-url")
215
-
xrpcProxyAdminPassword := cctx.String("xrpc-proxy-admin-password")
216
-
microNSFWImgURL := cctx.String("micro-nsfw-img-url")
217
-
hiveAIToken := cctx.String("hiveai-api-token")
218
-
sqrlURL := cctx.String("sqrl-url")
219
-
220
-
if repoPassword == "admin" {
221
-
log.Warn("using insecure default admin password (ok for dev, not for deployment)")
222
-
}
223
-
224
-
var serkey *did.PrivKey
225
-
if signingSecretKeyJwk != "" {
226
-
serkey, err = labeler.ParseSecretKey(signingSecretKeyJwk)
227
-
if err != nil {
228
-
return err
229
-
}
230
-
} else {
231
-
serkey, err = labeler.LoadOrCreateKeyFile(repoKeyPath, "auto-labelmaker")
232
-
if err != nil {
233
-
return err
234
-
}
235
-
}
236
-
237
-
repoUser := labeler.RepoConfig{
238
-
Handle: repoHandle,
239
-
Did: repoDid,
240
-
Password: repoPassword,
241
-
SigningKey: serkey,
242
-
UserId: 1,
243
-
}
244
-
245
-
srv, err := labeler.NewServer(db, cstore, repoUser, plcURL, blobPdsURL, xrpcProxyURL, xrpcProxyAdminPassword, useWss)
246
-
if err != nil {
247
-
return err
248
-
}
249
-
250
-
for _, l := range kwl {
251
-
srv.AddKeywordLabeler(l)
252
-
}
253
-
254
-
if microNSFWImgURL != "" {
255
-
srv.AddMicroNSFWImgLabeler(microNSFWImgURL)
256
-
}
257
-
258
-
if hiveAIToken != "" {
259
-
srv.AddHiveAILabeler(hiveAIToken)
260
-
}
261
-
262
-
if sqrlURL != "" {
263
-
srv.AddSQRLLabeler(sqrlURL)
264
-
}
265
-
266
-
srv.SubscribeBGS(context.TODO(), bgsURL, useWss)
267
-
return srv.RunAPI(bind)
268
-
}
269
-
270
-
return app.Run(args)
271
-
}
-3
cmd/labelmaker/sqrl_example/bsky/cryptocurrency_keywords.txt
-3
cmd/labelmaker/sqrl_example/bsky/cryptocurrency_keywords.txt
-7
cmd/labelmaker/sqrl_example/bsky/main.sqrl
-7
cmd/labelmaker/sqrl_example/bsky/main.sqrl
-14
cmd/labelmaker/sqrl_example/bsky/post.sqrl
-14
cmd/labelmaker/sqrl_example/bsky/post.sqrl
···
1
-
2
-
# EventData, EventType, UserDid already defined
3
-
LET Text := jsonValue(EventData, "$.post.text");
4
-
5
-
log("Text: %s", Text);
6
-
7
-
LET HasCryptoKeywords := patternMatches("cryptocurrency_keywords.txt", Text);
8
-
9
-
log("HasCryptoKeywords: %s", HasCryptoKeywords);
10
-
11
-
LET NumPostsAboutCrypto := count(BY UserDid WHERE HasCryptoKeywords LAST DAY);
12
-
13
-
CREATE RULE TooMuchCrypto WHERE NumPostsAboutCrypto > 5 WITH REASON "Repo ${UserDid} posted about crypto ${NumPostsAboutCrypto} times in the last day";
14
-
WHEN TooMuchCrypto THEN blockAction();
-3
cmd/labelmaker/sqrl_example/bsky/profile.sqrl
-3
cmd/labelmaker/sqrl_example/bsky/profile.sqrl
+7
-1
cmd/palomar/main.go
+7
-1
cmd/palomar/main.go
···
158
158
Value: 100,
159
159
EnvVars: []string{"PALOMAR_PLC_RATE_LIMIT"},
160
160
},
161
+
&cli.BoolFlag{
162
+
Name: "discover-repos",
163
+
Usage: "if true, discover repositories from the Relay",
164
+
EnvVars: []string{"PALOMAR_DISCOVER_REPOS"},
165
+
Value: false,
166
+
},
161
167
},
162
168
Action: func(cctx *cli.Context) error {
163
169
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
···
220
226
TryAuthoritativeDNS: true,
221
227
SkipDNSDomainSuffixes: []string{".bsky.social"},
222
228
}
223
-
dir := identity.NewCacheDirectory(&base, 1_500_000, time.Hour*24, time.Minute*2)
229
+
dir := identity.NewCacheDirectory(&base, 1_500_000, time.Hour*24, time.Minute*2, time.Minute*5)
224
230
225
231
srv, err := search.NewServer(
226
232
db,
-132
labeler/admin.go
-132
labeler/admin.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"context"
5
-
"fmt"
6
-
"time"
7
-
8
-
comatproto "github.com/bluesky-social/indigo/api/atproto"
9
-
appbsky "github.com/bluesky-social/indigo/api/bsky"
10
-
lexutil "github.com/bluesky-social/indigo/lex/util"
11
-
"github.com/bluesky-social/indigo/models"
12
-
)
13
-
14
-
// This is probably only a temporary method
15
-
func (s *Server) hydrateRepoView(ctx context.Context, did, indexedAt string) *comatproto.AdminDefs_RepoView {
16
-
return &comatproto.AdminDefs_RepoView{
17
-
// TODO(bnewbold): populate more, or more correctly, from some backend?
18
-
Did: did,
19
-
Email: nil,
20
-
Handle: "TODO",
21
-
IndexedAt: indexedAt,
22
-
Moderation: nil,
23
-
RelatedRecords: nil,
24
-
}
25
-
}
26
-
27
-
// This is probably only a temporary method
28
-
func (s *Server) hydrateRecordView(ctx context.Context, did string, uri, cid *string, indexedAt string) *comatproto.AdminDefs_RecordView {
29
-
repoView := s.hydrateRepoView(ctx, did, indexedAt)
30
-
// TODO(bnewbold): populate more, or more correctly, from some backend?
31
-
recordView := comatproto.AdminDefs_RecordView{
32
-
BlobCids: []string{},
33
-
IndexedAt: indexedAt,
34
-
Moderation: nil,
35
-
Repo: repoView,
36
-
// TODO: replace with actual record (from proxied backend)
37
-
Value: &lexutil.LexiconTypeDecoder{Val: &appbsky.FeedPost{}},
38
-
}
39
-
if uri != nil {
40
-
recordView.Uri = *uri
41
-
}
42
-
if cid != nil {
43
-
recordView.Cid = *cid
44
-
}
45
-
return &recordView
46
-
}
47
-
48
-
func (s *Server) hydrateModerationReportViews(ctx context.Context, rows []models.ModerationReport) ([]*comatproto.AdminDefs_ReportView, error) {
49
-
50
-
var out []*comatproto.AdminDefs_ReportView
51
-
for _, row := range rows {
52
-
var resolvedByActionIds []int64
53
-
var actionRows []models.ModerationAction
54
-
result := s.db.Joins("left join moderation_report_resolutions on moderation_report_resolutions.action_id = moderation_actions.id").Where("moderation_report_resolutions.report_id = ?", row.ID).Where("moderation_actions.reversed_at IS NULL").Find(&actionRows)
55
-
if result.Error != nil {
56
-
return nil, result.Error
57
-
}
58
-
for _, actionRow := range actionRows {
59
-
resolvedByActionIds = append(resolvedByActionIds, int64(actionRow.ID))
60
-
}
61
-
62
-
var subj *comatproto.AdminDefs_ReportView_Subject
63
-
switch row.SubjectType {
64
-
case "com.atproto.repo.repoRef":
65
-
subj = &comatproto.AdminDefs_ReportView_Subject{
66
-
AdminDefs_RepoRef: &comatproto.AdminDefs_RepoRef{
67
-
LexiconTypeID: "com.atproto.repo.repoRef",
68
-
Did: row.SubjectDid,
69
-
},
70
-
}
71
-
case "com.atproto.repo.recordRef":
72
-
subj = &comatproto.AdminDefs_ReportView_Subject{
73
-
RepoStrongRef: &comatproto.RepoStrongRef{
74
-
LexiconTypeID: "com.atproto.repo.strongRef",
75
-
Uri: *row.SubjectUri,
76
-
Cid: *row.SubjectCid,
77
-
},
78
-
}
79
-
default:
80
-
return nil, fmt.Errorf("unsupported moderation SubjectType: %v", row.SubjectType)
81
-
}
82
-
83
-
view := &comatproto.AdminDefs_ReportView{
84
-
Id: int64(row.ID),
85
-
Comment: row.Reason,
86
-
ReasonType: &row.ReasonType,
87
-
Subject: subj,
88
-
ReportedBy: row.ReportedByDid,
89
-
CreatedAt: row.CreatedAt.Format(time.RFC3339),
90
-
ResolvedByActionIds: resolvedByActionIds,
91
-
}
92
-
out = append(out, view)
93
-
}
94
-
return out, nil
95
-
}
96
-
97
-
func (s *Server) hydrateModerationReportDetails(ctx context.Context, rows []models.ModerationReport) ([]*comatproto.AdminDefs_ReportViewDetail, error) {
98
-
99
-
var out []*comatproto.AdminDefs_ReportViewDetail
100
-
for _, row := range rows {
101
-
var actionRows []models.ModerationAction
102
-
result := s.db.Joins("left join moderation_report_resolutions on moderation_report_resolutions.action_id = moderation_actions.id").Where("moderation_report_resolutions.report_id = ?", row.ID).Where("moderation_actions.reversed_at IS NULL").Find(&actionRows)
103
-
if result.Error != nil {
104
-
return nil, result.Error
105
-
}
106
-
107
-
var subj *comatproto.AdminDefs_ReportViewDetail_Subject
108
-
switch row.SubjectType {
109
-
case "com.atproto.repo.repoRef":
110
-
subj = &comatproto.AdminDefs_ReportViewDetail_Subject{
111
-
AdminDefs_RepoView: s.hydrateRepoView(ctx, row.SubjectDid, row.CreatedAt.Format(time.RFC3339)),
112
-
}
113
-
case "com.atproto.repo.recordRef":
114
-
subj = &comatproto.AdminDefs_ReportViewDetail_Subject{
115
-
AdminDefs_RecordView: s.hydrateRecordView(ctx, row.SubjectDid, row.SubjectUri, row.SubjectCid, row.CreatedAt.Format(time.RFC3339)),
116
-
}
117
-
default:
118
-
return nil, fmt.Errorf("unsupported moderation SubjectType: %v", row.SubjectType)
119
-
}
120
-
121
-
viewDetail := &comatproto.AdminDefs_ReportViewDetail{
122
-
Id: int64(row.ID),
123
-
Comment: row.Reason,
124
-
ReasonType: &row.ReasonType,
125
-
Subject: subj,
126
-
ReportedBy: row.ReportedByDid,
127
-
CreatedAt: row.CreatedAt.Format(time.RFC3339),
128
-
}
129
-
out = append(out, viewDetail)
130
-
}
131
-
return out, nil
132
-
}
-76
labeler/commit.go
-76
labeler/commit.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"context"
5
-
"fmt"
6
-
"strings"
7
-
"time"
8
-
9
-
comatproto "github.com/bluesky-social/indigo/api/atproto"
10
-
"github.com/bluesky-social/indigo/events"
11
-
"github.com/bluesky-social/indigo/models"
12
-
util "github.com/bluesky-social/indigo/util"
13
-
14
-
"gorm.io/gorm/clause"
15
-
)
16
-
17
-
// Persist to database (and repo), and emit events.
18
-
func (s *Server) CommitLabels(ctx context.Context, labels []*comatproto.LabelDefs_Label, negate bool) error {
19
-
20
-
now := time.Now()
21
-
nowStr := now.Format(util.ISO8601)
22
-
var labelRows []models.Label
23
-
24
-
for _, l := range labels {
25
-
l.Cts = nowStr
26
-
27
-
path, _, err := s.repoman.CreateRecord(ctx, s.user.UserId, "com.atproto.label.label", l)
28
-
if err != nil {
29
-
return fmt.Errorf("failed to persist label in local repo: %w", err)
30
-
}
31
-
labelUri := "at://" + s.user.Did + "/" + path
32
-
log.Infof("persisted label in repo: %s", labelUri)
33
-
34
-
rkey := strings.SplitN(path, "/", 2)[1]
35
-
lr := models.Label{
36
-
Uri: l.Uri,
37
-
SourceDid: l.Src,
38
-
Cid: l.Cid,
39
-
Val: l.Val,
40
-
Neg: nil,
41
-
RepoRKey: &rkey,
42
-
CreatedAt: now,
43
-
}
44
-
if negate {
45
-
t := true
46
-
lr.Neg = &t
47
-
}
48
-
labelRows = append(labelRows, lr)
49
-
}
50
-
51
-
// ... and database ...
52
-
if len(labelRows) > 0 {
53
-
// TODO(bnewbold): don't clobber action labels (aka, human interventions)
54
-
res := s.db.Clauses(clause.OnConflict{DoNothing: true}).Create(&labelRows)
55
-
if res.Error != nil {
56
-
return res.Error
57
-
}
58
-
}
59
-
60
-
// ... then re-publish as XRPCStreamEvent
61
-
if len(labels) > 0 {
62
-
log.Infof("broadcasting labels: %s", labels)
63
-
lev := events.XRPCStreamEvent{
64
-
LabelLabels: &comatproto.LabelSubscribeLabels_Labels{
65
-
// NOTE(bnewbold): generic event handler code handles Seq field for us
66
-
Labels: labels,
67
-
},
68
-
}
69
-
err := s.evtmgr.AddEvent(ctx, &lev)
70
-
if err != nil {
71
-
return fmt.Errorf("failed to publish XRPCStreamEvent: %w", err)
72
-
}
73
-
}
74
-
75
-
return nil
76
-
}
-99
labeler/helpers_test.go
-99
labeler/helpers_test.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"encoding/json"
5
-
"net/http"
6
-
"net/http/httptest"
7
-
"net/url"
8
-
"strconv"
9
-
"strings"
10
-
"testing"
11
-
12
-
"github.com/labstack/echo/v4"
13
-
"github.com/stretchr/testify/assert"
14
-
15
-
comatproto "github.com/bluesky-social/indigo/api/atproto"
16
-
)
17
-
18
-
// fetches report via getModerationReport, verifies match
19
-
func testGetReport(t *testing.T, e *echo.Echo, lm *Server, reportId int64) comatproto.AdminDefs_ReportViewDetail {
20
-
assert := assert.New(t)
21
-
22
-
params := make(url.Values)
23
-
params.Set("id", strconv.Itoa(int(reportId)))
24
-
req := httptest.NewRequest(http.MethodGet, "/xrpc/com.atproto.admin.getModerationReport?"+params.Encode(), nil)
25
-
recorder := httptest.NewRecorder()
26
-
c := e.NewContext(req, recorder)
27
-
assert.NoError(lm.HandleComAtprotoAdminGetModerationReport(c))
28
-
assert.Equal(200, recorder.Code)
29
-
var reportViewDetail comatproto.AdminDefs_ReportViewDetail
30
-
if err := json.Unmarshal([]byte(recorder.Body.String()), &reportViewDetail); err != nil {
31
-
t.Fatal(err)
32
-
}
33
-
assert.Equal(reportId, reportViewDetail.Id)
34
-
35
-
return reportViewDetail
36
-
}
37
-
38
-
// "happy path" test helper. creates a report, reads it back 2x ways, verifies match, then returns the original output
39
-
func testCreateReport(t *testing.T, e *echo.Echo, lm *Server, input *comatproto.ModerationCreateReport_Input) comatproto.ModerationCreateReport_Output {
40
-
assert := assert.New(t)
41
-
42
-
// create report and verify output
43
-
reportJSON, err := json.Marshal(input)
44
-
if err != nil {
45
-
t.Fatal(err)
46
-
}
47
-
req := httptest.NewRequest(http.MethodPost, "/xrpc/com.atproto.report.create", strings.NewReader(string(reportJSON)))
48
-
req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
49
-
recorder := httptest.NewRecorder()
50
-
c := e.NewContext(req, recorder)
51
-
52
-
assert.NoError(lm.HandleComAtprotoReportCreate(c))
53
-
assert.Equal(200, recorder.Code)
54
-
55
-
var out comatproto.ModerationCreateReport_Output
56
-
if err := json.Unmarshal([]byte(recorder.Body.String()), &out); err != nil {
57
-
t.Fatal(err)
58
-
}
59
-
assert.Equal(input.Reason, out.Reason)
60
-
assert.Equal(input.ReasonType, out.ReasonType)
61
-
assert.Equal(input.Subject.RepoStrongRef, out.Subject.RepoStrongRef)
62
-
assert.Equal(input.Subject.AdminDefs_RepoRef, out.Subject.AdminDefs_RepoRef)
63
-
64
-
// read it back and verify output
65
-
reportViewDetail := testGetReport(t, e, lm, out.Id)
66
-
assert.Equal(out.Id, reportViewDetail.Id)
67
-
assert.Equal(out.CreatedAt, reportViewDetail.CreatedAt)
68
-
assert.Equal(out.ReportedBy, reportViewDetail.ReportedBy)
69
-
assert.Equal(out.Reason, reportViewDetail.Comment)
70
-
assert.Equal(out.ReasonType, reportViewDetail.ReasonType)
71
-
assert.Equal(0, len(reportViewDetail.ResolvedByActions))
72
-
if out.Subject.AdminDefs_RepoRef != nil {
73
-
assert.Equal(out.Subject.AdminDefs_RepoRef.Did, reportViewDetail.Subject.AdminDefs_RepoView.Did)
74
-
} else if out.Subject.RepoStrongRef != nil {
75
-
assert.Equal(out.Subject.RepoStrongRef.Uri, reportViewDetail.Subject.AdminDefs_RecordView.Uri)
76
-
assert.Equal(out.Subject.RepoStrongRef.Cid, reportViewDetail.Subject.AdminDefs_RecordView.Cid)
77
-
} else {
78
-
t.Fatal("expected non-empty actionviewdetail.subject enum")
79
-
}
80
-
81
-
return out
82
-
}
83
-
84
-
func testQueryLabels(t *testing.T, e *echo.Echo, lm *Server, params *url.Values) (*comatproto.LabelQueryLabels_Output, error) {
85
-
86
-
req := httptest.NewRequest(http.MethodGet, "/xrpc/com.atproto.label.queryLabels?"+params.Encode(), nil)
87
-
req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
88
-
recorder := httptest.NewRecorder()
89
-
c := e.NewContext(req, recorder)
90
-
err := lm.HandleComAtprotoLabelQueryLabels(c)
91
-
if err != nil {
92
-
return nil, err
93
-
}
94
-
var out comatproto.LabelQueryLabels_Output
95
-
if err := json.Unmarshal([]byte(recorder.Body.String()), &out); err != nil {
96
-
t.Fatal(err)
97
-
}
98
-
return &out, nil
99
-
}
-146
labeler/hiveai.go
-146
labeler/hiveai.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"bytes"
5
-
"context"
6
-
"encoding/json"
7
-
"fmt"
8
-
"io"
9
-
"mime/multipart"
10
-
"net/http"
11
-
12
-
lexutil "github.com/bluesky-social/indigo/lex/util"
13
-
"github.com/bluesky-social/indigo/util"
14
-
15
-
"github.com/carlmjohnson/versioninfo"
16
-
)
17
-
18
-
type HiveAILabeler struct {
19
-
Client http.Client
20
-
ApiToken string
21
-
}
22
-
23
-
// schema: https://docs.thehive.ai/reference/classification
24
-
type HiveAIResp struct {
25
-
Status []HiveAIResp_Status `json:"status"`
26
-
}
27
-
28
-
type HiveAIResp_Status struct {
29
-
Response HiveAIResp_Response `json:"response"`
30
-
}
31
-
32
-
type HiveAIResp_Response struct {
33
-
Output []HiveAIResp_Out `json:"output"`
34
-
}
35
-
36
-
type HiveAIResp_Out struct {
37
-
Time float64 `json:"time"`
38
-
Classes []HiveAIResp_Class `json:"classes"`
39
-
}
40
-
41
-
type HiveAIResp_Class struct {
42
-
Class string `json:"class"`
43
-
Score float64 `json:"score"`
44
-
}
45
-
46
-
func NewHiveAILabeler(token string) HiveAILabeler {
47
-
return HiveAILabeler{
48
-
Client: *util.RobustHTTPClient(),
49
-
ApiToken: token,
50
-
}
51
-
}
52
-
53
-
func (resp *HiveAIResp) SummarizeLabels() []string {
54
-
var labels []string
55
-
56
-
for _, status := range resp.Status {
57
-
for _, out := range status.Response.Output {
58
-
for _, cls := range out.Classes {
59
-
// TODO(bnewbold): lots more upstream tags could be included here.
60
-
// for example, "sexy" for not nude but still explicit/suggestive,
61
-
// or lolicon (animated, not nude, "sugggestive"
62
-
63
-
// sexual: https://docs.thehive.ai/docs/sexual-content
64
-
// note: won't apply "nude" if "porn" already applied
65
-
if cls.Class == "yes_sexual_activity" && cls.Score >= 0.90 {
66
-
// NOTE: will include "hentai"
67
-
labels = append(labels, "porn")
68
-
} else if cls.Class == "animal_genitalia_and_human" && cls.Score >= 0.90 {
69
-
labels = append(labels, "porn")
70
-
} else if cls.Class == "yes_male_nudity" && cls.Score >= 0.90 {
71
-
labels = append(labels, "nude")
72
-
} else if cls.Class == "yes_female_nudity" && cls.Score >= 0.90 {
73
-
labels = append(labels, "nude")
74
-
}
75
-
76
-
// gore and violence: https://docs.thehive.ai/docs/class-descriptions-violence-gore
77
-
if cls.Class == "very_bloody" && cls.Score >= 0.90 {
78
-
labels = append(labels, "gore")
79
-
}
80
-
if cls.Class == "human_corpse" && cls.Score >= 0.90 {
81
-
labels = append(labels, "corpse")
82
-
}
83
-
if cls.Class == "yes_self_harm" && cls.Score >= 0.90 {
84
-
labels = append(labels, "self-harm")
85
-
}
86
-
}
87
-
}
88
-
}
89
-
90
-
return labels
91
-
}
92
-
93
-
func (hal *HiveAILabeler) LabelBlob(ctx context.Context, blob lexutil.LexBlob, blobBytes []byte) ([]string, error) {
94
-
95
-
log.Infof("sending blob to thehive.ai cid=%s mimetype=%s size=%d", blob.Ref, blob.MimeType, len(blobBytes))
96
-
97
-
// generic HTTP form file upload, then parse the response JSON
98
-
body := &bytes.Buffer{}
99
-
writer := multipart.NewWriter(body)
100
-
part, err := writer.CreateFormFile("media", blob.Ref.String())
101
-
if err != nil {
102
-
return nil, err
103
-
}
104
-
_, err = part.Write(blobBytes)
105
-
if err != nil {
106
-
return nil, err
107
-
}
108
-
err = writer.Close()
109
-
if err != nil {
110
-
return nil, err
111
-
}
112
-
113
-
req, err := http.NewRequest("POST", "https://api.thehive.ai/api/v2/task/sync", body)
114
-
if err != nil {
115
-
return nil, err
116
-
}
117
-
118
-
req.Header.Set("Authorization", fmt.Sprintf("Token %s", hal.ApiToken))
119
-
req.Header.Add("Content-Type", writer.FormDataContentType())
120
-
req.Header.Set("Accept", "application/json")
121
-
req.Header.Set("User-Agent", "labelmaker/"+versioninfo.Short())
122
-
123
-
res, err := hal.Client.Do(req)
124
-
if err != nil {
125
-
return nil, fmt.Errorf("HiveAI request failed: %v", err)
126
-
}
127
-
defer res.Body.Close()
128
-
if res.StatusCode != 200 {
129
-
return nil, fmt.Errorf("HiveAI request failed statusCode=%d", res.StatusCode)
130
-
}
131
-
132
-
respBytes, err := io.ReadAll(res.Body)
133
-
if err != nil {
134
-
return nil, fmt.Errorf("failed to read HiveAI resp body: %v", err)
135
-
}
136
-
137
-
log.Debugf("HiveAI raw result cid=%s body=%v", blob.Ref, string(respBytes))
138
-
139
-
var respObj HiveAIResp
140
-
if err := json.Unmarshal(respBytes, &respObj); err != nil {
141
-
return nil, fmt.Errorf("failed to parse HiveAI resp JSON: %v", err)
142
-
}
143
-
respJson, _ := json.Marshal(respObj.Status[0].Response.Output[0])
144
-
log.Infof("HiveAI result cid=%s json=%v", blob.Ref, string(respJson))
145
-
return respObj.SummarizeLabels(), nil
146
-
}
-42
labeler/hiveai_test.go
-42
labeler/hiveai_test.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"encoding/json"
5
-
"io"
6
-
"os"
7
-
"reflect"
8
-
"testing"
9
-
)
10
-
11
-
func TestHiveParse(t *testing.T) {
12
-
file, err := os.Open("testdata/hiveai_resp_example.json")
13
-
if err != nil {
14
-
t.Fatal(err)
15
-
}
16
-
17
-
respBytes, err := io.ReadAll(file)
18
-
if err != nil {
19
-
t.Fatal(err)
20
-
}
21
-
22
-
var respObj HiveAIResp
23
-
if err := json.Unmarshal(respBytes, &respObj); err != nil {
24
-
t.Fatal(err)
25
-
}
26
-
27
-
classes := respObj.Status[0].Response.Output[0].Classes
28
-
if len(classes) <= 10 {
29
-
t.Fatal("didn't get expected class count")
30
-
}
31
-
for _, c := range classes {
32
-
if c.Class == "" || c.Score == 0.0 {
33
-
t.Fatal("got null/empty class in resp")
34
-
}
35
-
}
36
-
37
-
labels := respObj.SummarizeLabels()
38
-
expected := []string{"porn"}
39
-
if !reflect.DeepEqual(labels, expected) {
40
-
t.Fatal("didn't summarize to expected labels")
41
-
}
42
-
}
-63
labeler/keyword_labeler.go
-63
labeler/keyword_labeler.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"encoding/json"
5
-
"fmt"
6
-
"io"
7
-
"os"
8
-
"strings"
9
-
10
-
appbsky "github.com/bluesky-social/indigo/api/bsky"
11
-
)
12
-
13
-
type KeywordLabeler struct {
14
-
Keywords []string `json:"keywords"`
15
-
Value string `json:"value"`
16
-
}
17
-
18
-
func (kl KeywordLabeler) LabelText(txt string) []string {
19
-
txt = strings.ToLower(txt)
20
-
for _, word := range kl.Keywords {
21
-
if strings.Contains(txt, word) {
22
-
return []string{kl.Value}
23
-
}
24
-
}
25
-
return []string{}
26
-
}
27
-
28
-
func (kl KeywordLabeler) LabelPost(p appbsky.FeedPost) []string {
29
-
return kl.LabelText(p.Text)
30
-
}
31
-
32
-
func (kl KeywordLabeler) LabelProfile(ap appbsky.ActorProfile) []string {
33
-
var txt string
34
-
if ap.DisplayName != nil {
35
-
txt += *ap.DisplayName
36
-
}
37
-
if ap.Description != nil {
38
-
txt += *ap.Description
39
-
}
40
-
return kl.LabelText(txt)
41
-
}
42
-
43
-
func LoadKeywordFile(fpath string) ([]KeywordLabeler, error) {
44
-
45
-
var kwl []KeywordLabeler
46
-
47
-
jsonFile, err := os.Open(fpath)
48
-
if err != nil {
49
-
return nil, fmt.Errorf("failed to load JSON file: %v", err)
50
-
}
51
-
defer jsonFile.Close()
52
-
53
-
raw, err := io.ReadAll(jsonFile)
54
-
if err != nil {
55
-
return nil, fmt.Errorf("failed to load JSON file: %v", err)
56
-
}
57
-
58
-
if err := json.Unmarshal(raw, &kwl); err != nil {
59
-
return nil, fmt.Errorf("failed to parse Keyword file: %v", err)
60
-
}
61
-
62
-
return kwl, nil
63
-
}
-53
labeler/keyword_labeler_test.go
-53
labeler/keyword_labeler_test.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"fmt"
5
-
"reflect"
6
-
"testing"
7
-
8
-
bsky "github.com/bluesky-social/indigo/api/bsky"
9
-
)
10
-
11
-
func TestKeywordFilter(t *testing.T) {
12
-
var kl = KeywordLabeler{Value: "rude", Keywords: []string{"🍆", "sex"}}
13
-
14
-
postCases := []struct {
15
-
record bsky.FeedPost
16
-
expected []string
17
-
}{
18
-
{bsky.FeedPost{Text: "boring inoffensive tweet"}, []string{}},
19
-
{bsky.FeedPost{Text: "I love Aubergine 🍆"}, []string{"rude"}},
20
-
{bsky.FeedPost{Text: "SeXyTiMe"}, []string{"rude"}},
21
-
}
22
-
23
-
for _, c := range postCases {
24
-
vals := kl.LabelPost(c.record)
25
-
if !reflect.DeepEqual(vals, c.expected) {
26
-
t.Log(fmt.Sprintf("labels expected:%s got:%s", c.expected, vals))
27
-
t.Fail()
28
-
}
29
-
}
30
-
31
-
desc := "yadda yadda"
32
-
descRude := "yadda yadda 🍆"
33
-
name := "Robyn Hood"
34
-
nameSexy := "Sexy Robyn Hood"
35
-
profileCases := []struct {
36
-
record bsky.ActorProfile
37
-
expected []string
38
-
}{
39
-
{bsky.ActorProfile{DisplayName: &name}, []string{}},
40
-
{bsky.ActorProfile{DisplayName: &name, Description: &desc}, []string{}},
41
-
{bsky.ActorProfile{DisplayName: &name, Description: &descRude}, []string{"rude"}},
42
-
{bsky.ActorProfile{DisplayName: &nameSexy}, []string{"rude"}},
43
-
{bsky.ActorProfile{DisplayName: &nameSexy, Description: &descRude}, []string{"rude"}},
44
-
}
45
-
46
-
for _, c := range profileCases {
47
-
vals := kl.LabelProfile(c.record)
48
-
if !reflect.DeepEqual(vals, c.expected) {
49
-
t.Log(fmt.Sprintf("labels expected:%s got:%s", c.expected, vals))
50
-
t.Fail()
51
-
}
52
-
}
53
-
}
-102
labeler/micro_nsfw_img.go
-102
labeler/micro_nsfw_img.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"bytes"
5
-
"context"
6
-
"encoding/json"
7
-
"fmt"
8
-
"io"
9
-
"mime/multipart"
10
-
"net/http"
11
-
12
-
lexutil "github.com/bluesky-social/indigo/lex/util"
13
-
util "github.com/bluesky-social/indigo/util"
14
-
15
-
"github.com/carlmjohnson/versioninfo"
16
-
)
17
-
18
-
type MicroNSFWImgLabeler struct {
19
-
Client http.Client
20
-
Endpoint string
21
-
}
22
-
23
-
type MicroNSFWImgResp struct {
24
-
Drawings float64 `json:"drawings"`
25
-
Hentai float64 `json:"hentai"`
26
-
Neutral float64 `json:"neutral"`
27
-
Porn float64 `json:"porn"`
28
-
Sexy float64 `json:"sexy"`
29
-
}
30
-
31
-
func NewMicroNSFWImgLabeler(url string) MicroNSFWImgLabeler {
32
-
return MicroNSFWImgLabeler{
33
-
Client: *util.RobustHTTPClient(),
34
-
Endpoint: url,
35
-
}
36
-
}
37
-
38
-
func (resp *MicroNSFWImgResp) SummarizeLabels() []string {
39
-
var labels []string
40
-
41
-
// TODO(bnewbold): these score cutoffs are kind of arbitrary
42
-
if resp.Porn > 0.90 {
43
-
labels = append(labels, "porn")
44
-
}
45
-
if resp.Hentai > 0.90 {
46
-
labels = append(labels, "hentai")
47
-
}
48
-
if resp.Sexy > 0.90 {
49
-
labels = append(labels, "sexy")
50
-
}
51
-
return labels
52
-
}
53
-
54
-
func (mnil *MicroNSFWImgLabeler) LabelBlob(ctx context.Context, blob lexutil.LexBlob, blobBytes []byte) ([]string, error) {
55
-
56
-
log.Infof("sending blob to micro-NSFW-img cid=%s mimetype=%s size=%d", blob.Ref, blob.MimeType, len(blobBytes))
57
-
58
-
// generic HTTP form file upload, then parse the response JSON
59
-
body := &bytes.Buffer{}
60
-
writer := multipart.NewWriter(body)
61
-
part, err := writer.CreateFormFile("file", blob.Ref.String())
62
-
if err != nil {
63
-
return nil, err
64
-
}
65
-
66
-
_, err = part.Write(blobBytes)
67
-
if err != nil {
68
-
return nil, err
69
-
}
70
-
err = writer.Close()
71
-
if err != nil {
72
-
return nil, err
73
-
}
74
-
req, err := http.NewRequest("POST", mnil.Endpoint, body)
75
-
if err != nil {
76
-
return nil, err
77
-
}
78
-
req.Header.Add("Content-Type", writer.FormDataContentType())
79
-
req.Header.Set("User-Agent", "labelmaker/"+versioninfo.Short())
80
-
81
-
res, err := mnil.Client.Do(req)
82
-
if err != nil {
83
-
return nil, fmt.Errorf("micro-NSFW-img request failed: %v", err)
84
-
}
85
-
defer res.Body.Close()
86
-
if res.StatusCode != 200 {
87
-
return nil, fmt.Errorf("micro-NSFW-img request failed statusCode=%d", res.StatusCode)
88
-
}
89
-
90
-
respBytes, err := io.ReadAll(res.Body)
91
-
if err != nil {
92
-
return nil, fmt.Errorf("failed to read micro-NSFW-img resp body: %v", err)
93
-
}
94
-
95
-
var nsfwScore MicroNSFWImgResp
96
-
if err := json.Unmarshal(respBytes, &nsfwScore); err != nil {
97
-
return nil, fmt.Errorf("failed to parse micro-NSFW-img resp JSON: %v", err)
98
-
}
99
-
scoreJson, _ := json.Marshal(nsfwScore)
100
-
log.Infof("micro-NSFW-img result cid=%s scores=%v", blob.Ref, string(scoreJson))
101
-
return nsfwScore.SummarizeLabels(), nil
102
-
}
-480
labeler/service.go
-480
labeler/service.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"bytes"
5
-
"context"
6
-
"crypto/subtle"
7
-
"encoding/base64"
8
-
"fmt"
9
-
"io"
10
-
"net/http"
11
-
"net/url"
12
-
"strings"
13
-
14
-
"github.com/bluesky-social/indigo/api"
15
-
comatproto "github.com/bluesky-social/indigo/api/atproto"
16
-
appbsky "github.com/bluesky-social/indigo/api/bsky"
17
-
"github.com/bluesky-social/indigo/bgs"
18
-
"github.com/bluesky-social/indigo/carstore"
19
-
"github.com/bluesky-social/indigo/events"
20
-
"github.com/bluesky-social/indigo/indexer"
21
-
lexutil "github.com/bluesky-social/indigo/lex/util"
22
-
"github.com/bluesky-social/indigo/models"
23
-
"github.com/bluesky-social/indigo/repo"
24
-
"github.com/bluesky-social/indigo/repomgr"
25
-
cbg "github.com/whyrusleeping/cbor-gen"
26
-
27
-
logging "github.com/ipfs/go-log"
28
-
"github.com/labstack/echo/v4"
29
-
"github.com/labstack/echo/v4/middleware"
30
-
"github.com/whyrusleeping/go-did"
31
-
"gorm.io/gorm"
32
-
)
33
-
34
-
var log = logging.Logger("labelmaker")
35
-
36
-
type Server struct {
37
-
db *gorm.DB
38
-
cs *carstore.CarStore
39
-
repoman *repomgr.RepoManager
40
-
bgsSlurper *bgs.Slurper
41
-
evtmgr *events.EventManager
42
-
echo *echo.Echo
43
-
user *RepoConfig
44
-
blobPdsURL string
45
-
xrpcProxyURL *url.URL
46
-
xrpcProxyAuthHeader string
47
-
kwLabelers []KeywordLabeler
48
-
muNSFWImgLabeler *MicroNSFWImgLabeler
49
-
hiveAILabeler *HiveAILabeler
50
-
sqrlLabeler *SQRLLabeler
51
-
}
52
-
53
-
type RepoConfig struct {
54
-
Handle string
55
-
Did string
56
-
Password string
57
-
SigningKey *did.PrivKey
58
-
UserId models.Uid
59
-
}
60
-
61
-
// In addition to configuring the service, will connect to upstream BGS and start processing events. Won't handle HTTP or WebSocket endpoints until RunAPI() is called.
62
-
// 'useWss' is a flag to use SSL for outbound WebSocket connections
63
-
func NewServer(db *gorm.DB, cs *carstore.CarStore, repoUser RepoConfig, plcURL, blobPdsURL, xrpcProxyURL, xrpcProxyAdminPassword string, useWss bool) (*Server, error) {
64
-
65
-
db.AutoMigrate(models.PDS{})
66
-
db.AutoMigrate(models.Label{})
67
-
db.AutoMigrate(models.ModerationAction{})
68
-
db.AutoMigrate(models.ModerationActionSubjectBlobCid{})
69
-
db.AutoMigrate(models.ModerationReport{})
70
-
db.AutoMigrate(models.ModerationReportResolution{})
71
-
72
-
didr := &api.PLCServer{Host: plcURL}
73
-
kmgr := indexer.NewKeyManager(didr, repoUser.SigningKey)
74
-
evtmgr := events.NewEventManager(events.NewMemPersister())
75
-
repoman := repomgr.NewRepoManager(cs, kmgr)
76
-
77
-
if repoUser.Password == "" || repoUser.Did == "" || repoUser.Handle == "" {
78
-
return nil, fmt.Errorf("bad labeler repo config (empty string)")
79
-
}
80
-
81
-
proxyURL, err := url.ParseRequestURI(xrpcProxyURL)
82
-
if err != nil {
83
-
return nil, fmt.Errorf("could not parse XRPC proxy URL (%v): %v", xrpcProxyURL, err)
84
-
}
85
-
xrpcProxyAuthHeader := "Basic " + base64.StdEncoding.EncodeToString([]byte("admin:"+xrpcProxyAdminPassword))
86
-
87
-
s := &Server{
88
-
db: db,
89
-
repoman: repoman,
90
-
evtmgr: evtmgr,
91
-
user: &repoUser,
92
-
blobPdsURL: blobPdsURL,
93
-
xrpcProxyURL: proxyURL,
94
-
xrpcProxyAuthHeader: xrpcProxyAuthHeader,
95
-
// sluper configured below
96
-
}
97
-
98
-
// ensure that local labelmaker repo exists
99
-
// NOTE: doesn't need to have app.bsky profile and actor config, this is just expediant (reusing an existing helper function)
100
-
ctx := context.Background()
101
-
head, _ := s.repoman.GetRepoRoot(ctx, s.user.UserId)
102
-
if !head.Defined() {
103
-
log.Info("initializing labelmaker repo")
104
-
if err := s.repoman.InitNewActor(ctx, s.user.UserId, s.user.Handle, s.user.Did, "Label Maker", "", ""); err != nil {
105
-
return nil, fmt.Errorf("creating labelmaker repo: %w", err)
106
-
}
107
-
} else {
108
-
log.Infof("found labelmaker repo: %s", head)
109
-
}
110
-
111
-
slOpts := bgs.DefaultSlurperOptions()
112
-
slOpts.SSL = useWss
113
-
slurp, err := bgs.NewSlurper(db, s.handleBgsRepoEvent, slOpts)
114
-
if err != nil {
115
-
return nil, err
116
-
}
117
-
s.bgsSlurper = slurp
118
-
119
-
return s, nil
120
-
}
121
-
122
-
func (s *Server) AddKeywordLabeler(kwl KeywordLabeler) {
123
-
log.Infof("configuring keyword labeler")
124
-
s.kwLabelers = append(s.kwLabelers, kwl)
125
-
}
126
-
127
-
func (s *Server) AddMicroNSFWImgLabeler(url string) {
128
-
log.Infof("configuring micro-NSFW-img labeler url=%s", url)
129
-
mnil := NewMicroNSFWImgLabeler(url)
130
-
s.muNSFWImgLabeler = &mnil
131
-
}
132
-
133
-
func (s *Server) AddHiveAILabeler(apiToken string) {
134
-
log.Infof("configuring Hive AI labeler")
135
-
hal := NewHiveAILabeler(apiToken)
136
-
s.hiveAILabeler = &hal
137
-
}
138
-
139
-
func (s *Server) AddSQRLLabeler(url string) {
140
-
log.Infof("configuring SQRL labeler url=%s", url)
141
-
sl := NewSQRLLabeler(url)
142
-
s.sqrlLabeler = &sl
143
-
}
144
-
145
-
// call this *after* all the labelers are configured
146
-
func (s *Server) SubscribeBGS(ctx context.Context, bgsURL string, useWss bool) {
147
-
// subscribe our RepoEvent slurper to the BGS, to receive incoming records for labeler
148
-
log.Infof("subscribing to BGS: %s (SSL=%v)", bgsURL, useWss)
149
-
s.bgsSlurper.SubscribeToPds(ctx, bgsURL, useWss, false)
150
-
}
151
-
152
-
// efficiency predicate to quickly discard events we know that we shouldn't even bother parsing
153
-
func (s *Server) wantAnyRecords(ctx context.Context, ra *comatproto.SyncSubscribeRepos_Commit) bool {
154
-
155
-
for _, op := range ra.Ops {
156
-
if op.Action != "create" && op.Action != "update" {
157
-
continue
158
-
}
159
-
nsid := strings.SplitN(op.Path, "/", 2)[0]
160
-
switch nsid {
161
-
case "app.bsky.feed.post":
162
-
return true
163
-
case "app.bsky.actor.profile":
164
-
return true
165
-
default:
166
-
continue
167
-
}
168
-
}
169
-
return false
170
-
}
171
-
172
-
// should we bother to fetch blob for processing?
173
-
func (s *Server) wantBlob(ctx context.Context, blob *lexutil.LexBlob) bool {
174
-
log.Debugf("wantBlob blob=%v", blob)
175
-
// images
176
-
if blob.MimeType == "image/png" || blob.MimeType == "image/jpeg" {
177
-
// only an image API is configured
178
-
if s.muNSFWImgLabeler != nil || s.hiveAILabeler != nil {
179
-
return true
180
-
}
181
-
}
182
-
return false
183
-
}
184
-
185
-
func (s *Server) labelRecord(ctx context.Context, did, nsid, uri, cidStr string, rec cbg.CBORMarshaler) ([]string, error) {
186
-
log.Infof("labeling record: %v", uri)
187
-
var labelVals []string
188
-
var blobs []lexutil.LexBlob
189
-
switch nsid {
190
-
case "app.bsky.feed.post":
191
-
post, suc := rec.(*appbsky.FeedPost)
192
-
if !suc {
193
-
return nil, fmt.Errorf("record failed to deserialize from CBOR: %s", rec)
194
-
}
195
-
196
-
// run through all the keyword labelers on posts, saving any resulting labels
197
-
for _, labeler := range s.kwLabelers {
198
-
for _, val := range labeler.LabelPost(*post) {
199
-
labelVals = append(labelVals, val)
200
-
}
201
-
}
202
-
203
-
if s.sqrlLabeler != nil {
204
-
sqrlVals, err := s.sqrlLabeler.LabelPost(ctx, *post)
205
-
if err != nil {
206
-
return nil, fmt.Errorf("failed to label post with SQRL: %v", err)
207
-
}
208
-
labelVals = append(labelVals, sqrlVals...)
209
-
}
210
-
211
-
// record any image blobs for processing
212
-
if post.Embed != nil && post.Embed.EmbedImages != nil {
213
-
for _, eii := range post.Embed.EmbedImages.Images {
214
-
blobs = append(blobs, *eii.Image)
215
-
}
216
-
}
217
-
case "app.bsky.actor.profile":
218
-
profile, suc := rec.(*appbsky.ActorProfile)
219
-
if !suc {
220
-
return nil, fmt.Errorf("record failed to deserialize from CBOR: %s", rec)
221
-
}
222
-
223
-
// run through all the keyword labelers on posts, saving any resulting labels
224
-
for _, labeler := range s.kwLabelers {
225
-
for _, val := range labeler.LabelProfile(*profile) {
226
-
labelVals = append(labelVals, val)
227
-
}
228
-
}
229
-
230
-
if s.sqrlLabeler != nil {
231
-
sqrlVals, err := s.sqrlLabeler.LabelProfile(ctx, *profile)
232
-
if err != nil {
233
-
return nil, fmt.Errorf("failed to label profile with SQRL: %v", err)
234
-
}
235
-
labelVals = append(labelVals, sqrlVals...)
236
-
}
237
-
238
-
// record avatar and/or banner blobs for processing
239
-
if profile.Avatar != nil {
240
-
blobs = append(blobs, *profile.Avatar)
241
-
}
242
-
if profile.Banner != nil {
243
-
blobs = append(blobs, *profile.Banner)
244
-
}
245
-
}
246
-
247
-
log.Infof("will process %d blobs", len(blobs))
248
-
for _, blob := range blobs {
249
-
if !blob.Ref.Defined() {
250
-
return nil, fmt.Errorf("received stub blob (CID undefined)")
251
-
}
252
-
253
-
if !s.wantBlob(ctx, &blob) {
254
-
log.Infof("skipping blob: cid=%s", blob.Ref.String())
255
-
continue
256
-
}
257
-
// download image for process
258
-
blobBytes, err := s.downloadRepoBlob(ctx, did, &blob)
259
-
// TODO(bnewbold): instead of erroring, just log any download problems
260
-
if err != nil {
261
-
return nil, err
262
-
}
263
-
264
-
blobLabels, err := s.labelBlob(ctx, did, blob, blobBytes)
265
-
// TODO(bnewbold): again, instead of erroring, just log any download problems
266
-
if err != nil {
267
-
return nil, err
268
-
}
269
-
labelVals = append(labelVals, blobLabels...)
270
-
}
271
-
return dedupeStrings(labelVals), nil
272
-
}
273
-
274
-
func (s *Server) downloadRepoBlob(ctx context.Context, did string, blob *lexutil.LexBlob) ([]byte, error) {
275
-
var blobBytes []byte
276
-
277
-
if !blob.Ref.Defined() {
278
-
return nil, fmt.Errorf("invalid blob to download (CID undefined)")
279
-
}
280
-
281
-
log.Infof("downloading blob pds=%s did=%s cid=%s", s.blobPdsURL, did, blob.Ref.String())
282
-
283
-
// TODO(bnewbold): more robust blob fetch code, by constructing query param
284
-
// properly; looking up DID doc; using xrpc.Client (with persistend HTTP
285
-
// client); etc.
286
-
// blocked on getBlob atproto branch landing, with new Lexicon.
287
-
// for now, just fetching from configured PDS (aka our single PDS)
288
-
xrpcURL := fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s", s.blobPdsURL, did, blob.Ref.String())
289
-
290
-
resp, err := http.Get(xrpcURL)
291
-
if err != nil {
292
-
return nil, err
293
-
}
294
-
defer resp.Body.Close()
295
-
296
-
if resp.StatusCode != 200 {
297
-
return nil, fmt.Errorf("failed to fetch blob from PDS. did=%s cid=%s statusCode=%d", did, blob.Ref.String(), resp.StatusCode)
298
-
}
299
-
300
-
blobBytes, err = io.ReadAll(resp.Body)
301
-
if err != nil {
302
-
return nil, err
303
-
}
304
-
305
-
return blobBytes, nil
306
-
}
307
-
308
-
func (s *Server) labelBlob(ctx context.Context, did string, blob lexutil.LexBlob, blobBytes []byte) ([]string, error) {
309
-
310
-
var labelVals []string
311
-
312
-
if !blob.Ref.Defined() {
313
-
return nil, fmt.Errorf("invalid blob to label (CID undefined)")
314
-
}
315
-
316
-
if s.muNSFWImgLabeler != nil {
317
-
318
-
nsfwLabels, err := s.muNSFWImgLabeler.LabelBlob(ctx, blob, blobBytes)
319
-
if err != nil {
320
-
return nil, err
321
-
}
322
-
labelVals = append(labelVals, nsfwLabels...)
323
-
}
324
-
325
-
if s.hiveAILabeler != nil {
326
-
327
-
hiveLabels, err := s.hiveAILabeler.LabelBlob(ctx, blob, blobBytes)
328
-
if err != nil {
329
-
return nil, err
330
-
}
331
-
labelVals = append(labelVals, hiveLabels...)
332
-
}
333
-
334
-
return labelVals, nil
335
-
}
336
-
337
-
// Process incoming repo events coming from BGS, which includes new and updated
338
-
// records from any PDS. This function extracts records, handes them to the
339
-
// labeling routine, and then persists and broadcasts any resulting labels
340
-
func (s *Server) handleBgsRepoEvent(ctx context.Context, pds *models.PDS, evt *events.XRPCStreamEvent) error {
341
-
342
-
if evt.RepoCommit == nil {
343
-
// TODO(bnewbold): is this really invalid? do we need to handle Info and Error events here?
344
-
return fmt.Errorf("invalid repo commit event")
345
-
}
346
-
347
-
// quick check if we can skip processing the CAR slice entirely
348
-
if !s.wantAnyRecords(ctx, evt.RepoCommit) {
349
-
return nil
350
-
}
351
-
352
-
// use an in-memory blockstore with repo wrapper to parse CAR slice
353
-
sliceRepo, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(evt.RepoCommit.Blocks))
354
-
if err != nil {
355
-
log.Warnw("failed to parse CAR slice", "repoErr", err)
356
-
return err
357
-
}
358
-
359
-
labels := []*comatproto.LabelDefs_Label{}
360
-
for _, op := range evt.RepoCommit.Ops {
361
-
uri := "at://" + evt.RepoCommit.Repo + "/" + op.Path
362
-
nsid := strings.SplitN(op.Path, "/", 2)[0]
363
-
364
-
if !(op.Action == "create" || op.Action == "update") {
365
-
continue
366
-
}
367
-
368
-
cid, rec, err := sliceRepo.GetRecord(ctx, op.Path)
369
-
if err != nil {
370
-
return fmt.Errorf("record not in CAR slice: %s", uri)
371
-
}
372
-
cidStr := cid.String()
373
-
labelVals, err := s.labelRecord(ctx, evt.RepoCommit.Repo, nsid, uri, cidStr, rec)
374
-
if err != nil {
375
-
return err
376
-
}
377
-
for _, val := range labelVals {
378
-
// apply labels with this pattern to the whole repo, not the record
379
-
if strings.HasPrefix(val, "repo:") {
380
-
val = strings.SplitN(val, ":", 2)[1]
381
-
labels = append(labels, &comatproto.LabelDefs_Label{
382
-
Src: s.user.Did,
383
-
Uri: "at://" + evt.RepoCommit.Repo,
384
-
Val: val,
385
-
//Neg
386
-
//Cts
387
-
})
388
-
} else {
389
-
labels = append(labels, &comatproto.LabelDefs_Label{
390
-
Src: s.user.Did,
391
-
Uri: uri,
392
-
Cid: &cidStr,
393
-
Val: val,
394
-
//Neg
395
-
//Cts
396
-
})
397
-
}
398
-
}
399
-
}
400
-
401
-
// persist and emit events, as needed
402
-
if err := s.CommitLabels(ctx, labels, false); err != nil {
403
-
return err
404
-
}
405
-
406
-
// TODO(bnewbold): persist state that we successfully processed the repo event (aka,
407
-
// persist "last" seq in database, or something like that). also above, at
408
-
// the short-circuit
409
-
return nil
410
-
}
411
-
412
-
// crude auth middleware to require "admin token" authentication on a subset of
413
-
// routes. Does not implement the usual atproto JWT-based auth. "admin token"
414
-
// auth is just HTTP Basic auth with username "admin" and a static password.
415
-
// TODO: either transition to some other auth scheme, or review this more carefully
416
-
func (s *Server) adminAuthMiddleware() echo.MiddlewareFunc {
417
-
config := middleware.BasicAuthConfig{
418
-
Skipper: func(c echo.Context) bool {
419
-
path := c.Request().URL.Path
420
-
// all admin paths require auth
421
-
if strings.HasPrefix(path, "/xrpc/com.atproto.admin.") {
422
-
return false
423
-
}
424
-
// TODO: will need more complex auth on this endpoint eventually
425
-
if strings.HasPrefix(path, "/xrpc/com.atproto.report.create") {
426
-
return false
427
-
}
428
-
// everything else defaults open
429
-
return true
430
-
},
431
-
Validator: func(username, password string, c echo.Context) (bool, error) {
432
-
// this is the default HTTP Basic validator from echo docs
433
-
// "Be careful to use constant time comparison to prevent timing attacks"
434
-
if subtle.ConstantTimeCompare([]byte(username), []byte("admin")) == 1 &&
435
-
subtle.ConstantTimeCompare([]byte(password), []byte(s.user.Password)) == 1 {
436
-
return true, nil
437
-
}
438
-
log.Warnw("auth failed", "username", string(username))
439
-
return false, nil
440
-
},
441
-
Realm: "AtprotoLabeler",
442
-
}
443
-
return middleware.BasicAuthWithConfig(config)
444
-
}
445
-
446
-
func (s *Server) RunAPI(listen string) error {
447
-
e := echo.New()
448
-
s.echo = e
449
-
e.HideBanner = true
450
-
e.Use(middleware.LoggerWithConfig(middleware.LoggerConfig{
451
-
Format: "method=${method} uri=${uri} status=${status} latency=${latency_human}\n",
452
-
}))
453
-
e.Use(s.adminAuthMiddleware())
454
-
455
-
e.HTTPErrorHandler = func(err error, ctx echo.Context) {
456
-
code := 500
457
-
if he, ok := err.(*echo.HTTPError); ok {
458
-
code = he.Code
459
-
}
460
-
log.Warnw("HTTP request error", "statusCode", code, "path", ctx.Path(), "err", err)
461
-
ctx.Response().WriteHeader(code)
462
-
}
463
-
464
-
e.GET("/xrpc/_health", s.HandleHealthCheck)
465
-
if err := s.RegisterHandlersComAtproto(e); err != nil {
466
-
return err
467
-
}
468
-
if err := s.RegisterProxyHandlers(e); err != nil {
469
-
return err
470
-
}
471
-
// single websocket endpoint
472
-
e.GET("/xrpc/com.atproto.label.subscribeLabels", s.EventsLabelsWebsocket)
473
-
474
-
log.Infof("starting labelmaker XRPC and WebSocket daemon at: %s", listen)
475
-
return e.Start(listen)
476
-
}
477
-
478
-
func (s *Server) Shutdown(ctx context.Context) error {
479
-
return s.echo.Shutdown(ctx)
480
-
}
-62
labeler/service_test.go
-62
labeler/service_test.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"os"
5
-
"path/filepath"
6
-
"testing"
7
-
8
-
"github.com/bluesky-social/indigo/carstore"
9
-
"gorm.io/driver/sqlite"
10
-
"gorm.io/gorm"
11
-
)
12
-
13
-
func testLabelMaker(t *testing.T) *Server {
14
-
15
-
tempdir, err := os.MkdirTemp("", "labelmaker-test-")
16
-
if err != nil {
17
-
t.Fatal(err)
18
-
}
19
-
sharddir := filepath.Join(tempdir, "shards")
20
-
if err := os.MkdirAll(sharddir, 0775); err != nil {
21
-
t.Fatal(err)
22
-
}
23
-
24
-
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{SkipDefaultTransaction: true})
25
-
if err != nil {
26
-
t.Fatal(err)
27
-
}
28
-
29
-
cs, err := carstore.NewCarStore(db, sharddir)
30
-
if err != nil {
31
-
t.Fatal(err)
32
-
}
33
-
34
-
repoKeyPath := filepath.Join(tempdir, "labelmaker.key")
35
-
serkey, err := LoadOrCreateKeyFile(repoKeyPath, "auto-labelmaker")
36
-
if err != nil {
37
-
t.Fatal(err)
38
-
}
39
-
40
-
plcURL := "http://did-plc-test.dummy"
41
-
blobPdsURL := "http://pds-test.dummy"
42
-
xrpcProxyURL := "http://pds-test.dummy"
43
-
xrpcProxyAdminPassword := "xrpc-test-password"
44
-
repoUser := RepoConfig{
45
-
Handle: "test.handle.dummy",
46
-
Did: "did:plc:testdummy",
47
-
Password: "admin-test-password",
48
-
SigningKey: serkey,
49
-
UserId: 1,
50
-
}
51
-
52
-
lm, err := NewServer(db, cs, repoUser, plcURL, blobPdsURL, xrpcProxyURL, xrpcProxyAdminPassword, false)
53
-
if err != nil {
54
-
t.Fatal(err)
55
-
}
56
-
return lm
57
-
}
58
-
59
-
func TestLabelMakerCreation(t *testing.T) {
60
-
lm := testLabelMaker(t)
61
-
_ = lm
62
-
}
-129
labeler/sqrl.go
-129
labeler/sqrl.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"bytes"
5
-
"context"
6
-
"encoding/json"
7
-
"fmt"
8
-
"io"
9
-
"net/http"
10
-
11
-
appbsky "github.com/bluesky-social/indigo/api/bsky"
12
-
"github.com/bluesky-social/indigo/util"
13
-
14
-
"github.com/carlmjohnson/versioninfo"
15
-
)
16
-
17
-
type SQRLLabeler struct {
18
-
Client http.Client
19
-
Endpoint string
20
-
}
21
-
22
-
type SQRLRequest struct {
23
-
Type string `json:"type"`
24
-
Post *appbsky.FeedPost `json:"post"`
25
-
Profile *appbsky.ActorProfile `json:"profile"`
26
-
}
27
-
28
-
type SQRLRequest_Wrap struct {
29
-
EventData SQRLRequest `json:"EventData"`
30
-
}
31
-
32
-
type SQRLResponse struct {
33
-
Allow bool `json:"allow"`
34
-
Verdict SQRLResponse_Verdict `json:"verdict"`
35
-
Rules map[string]SQRLResponse_Rule `json:"rules"`
36
-
Features map[string]any `json:"features"`
37
-
}
38
-
39
-
type SQRLResponse_Verdict struct {
40
-
BlockRules []string `json:"blockRules"`
41
-
WhitelistRules []string `json:"whitelistRules"`
42
-
}
43
-
44
-
type SQRLResponse_Rule struct {
45
-
Reason string `json:"reason"`
46
-
}
47
-
48
-
func NewSQRLLabeler(url string) SQRLLabeler {
49
-
return SQRLLabeler{
50
-
Client: *util.RobustHTTPClient(),
51
-
Endpoint: url,
52
-
}
53
-
}
54
-
55
-
func (sl *SQRLLabeler) submitEvent(sqlrReq SQRLRequest) (*SQRLResponse, error) {
56
-
57
-
wrapped := SQRLRequest_Wrap{EventData: sqlrReq}
58
-
bodyJson, err := json.Marshal(wrapped)
59
-
if err != nil {
60
-
return nil, err
61
-
}
62
-
63
-
req, err := http.NewRequest("POST", sl.Endpoint+"?features=EventType", bytes.NewBuffer(bodyJson))
64
-
if err != nil {
65
-
return nil, err
66
-
}
67
-
68
-
req.Header.Add("Content-Type", "application/json")
69
-
req.Header.Set("Accept", "application/json")
70
-
req.Header.Set("User-Agent", "labelmaker/"+versioninfo.Short())
71
-
72
-
res, err := sl.Client.Do(req)
73
-
if err != nil {
74
-
return nil, fmt.Errorf("SQRL request failed: %v", err)
75
-
}
76
-
defer res.Body.Close()
77
-
if res.StatusCode != 200 {
78
-
return nil, fmt.Errorf("SQRL request failed statusCode=%d", res.StatusCode)
79
-
}
80
-
81
-
respBytes, err := io.ReadAll(res.Body)
82
-
if err != nil {
83
-
return nil, fmt.Errorf("failed to read SQRL resp body: %v", err)
84
-
}
85
-
86
-
var respObj SQRLResponse
87
-
if err := json.Unmarshal(respBytes, &respObj); err != nil {
88
-
return nil, fmt.Errorf("failed to parse SQRL resp JSON: %v", err)
89
-
}
90
-
respJson, _ := json.Marshal(respObj)
91
-
log.Infof("SQRL result json=%v", string(respJson))
92
-
return &respObj, nil
93
-
}
94
-
95
-
func (sl *SQRLLabeler) LabelPost(ctx context.Context, post appbsky.FeedPost) ([]string, error) {
96
-
var labels []string
97
-
req := SQRLRequest{
98
-
Type: "post",
99
-
Post: &post,
100
-
}
101
-
resp, err := sl.submitEvent(req)
102
-
if err != nil {
103
-
return nil, err
104
-
}
105
-
for name, _ := range resp.Rules {
106
-
if name == "TooMuchCrypto" {
107
-
labels = append(labels, "repo:crypto-shill")
108
-
}
109
-
}
110
-
return labels, nil
111
-
}
112
-
113
-
func (sl *SQRLLabeler) LabelProfile(ctx context.Context, profile appbsky.ActorProfile) ([]string, error) {
114
-
var labels []string
115
-
req := SQRLRequest{
116
-
Type: "profile",
117
-
Profile: &profile,
118
-
}
119
-
resp, err := sl.submitEvent(req)
120
-
if err != nil {
121
-
return nil, err
122
-
}
123
-
for name, _ := range resp.Rules {
124
-
if name == "TooMuchCrypto" {
125
-
labels = append(labels, "repo:crypto-shill")
126
-
}
127
-
}
128
-
return labels, nil
129
-
}
-401
labeler/testdata/hiveai_resp_example.json
-401
labeler/testdata/hiveai_resp_example.json
···
1
-
{
2
-
"id": "02122580-c37f-11ed-81d2-000000000000",
3
-
"code": 200,
4
-
"project_id": 12345,
5
-
"user_id": 12345,
6
-
"created_on": "2023-03-15T22:16:18.408Z",
7
-
"status": [
8
-
{
9
-
"status": {
10
-
"code": "0",
11
-
"message": "SUCCESS"
12
-
},
13
-
"response": {
14
-
"input": {
15
-
"id": "02122580-c37f-11ed-81d2-000000000000",
16
-
"charge": 0.003,
17
-
"model": "mod55_dense",
18
-
"model_version": 1,
19
-
"model_type": "CATEGORIZATION",
20
-
"created_on": "2023-03-15T22:16:18.136Z",
21
-
"media": {
22
-
"url": null,
23
-
"filename": "bafkreiam7k6mvkyuoybq4ynhljvj5xa75sdbhjbolzjf5j2udx7vj5gnsy",
24
-
"type": "PHOTO",
25
-
"mime_type": "jpeg",
26
-
"mimetype": "image/jpeg",
27
-
"width": 800,
28
-
"height": 800,
29
-
"num_frames": 1,
30
-
"duration": 0
31
-
},
32
-
"user_id": 12345,
33
-
"project_id": 12345,
34
-
"config_version": 1,
35
-
"config_tag": "default"
36
-
},
37
-
"output": [
38
-
{
39
-
"time": 0,
40
-
"classes": [
41
-
{
42
-
"class": "general_not_nsfw_not_suggestive",
43
-
"score": 0.9998097218132356
44
-
},
45
-
{
46
-
"class": "general_nsfw",
47
-
"score": 8.857344804177162e-05
48
-
},
49
-
{
50
-
"class": "general_suggestive",
51
-
"score": 0.00010170473872266839
52
-
},
53
-
{
54
-
"class": "no_female_underwear",
55
-
"score": 0.9999923079040384
56
-
},
57
-
{
58
-
"class": "yes_female_underwear",
59
-
"score": 7.692095961599136e-06
60
-
},
61
-
{
62
-
"class": "no_male_underwear",
63
-
"score": 0.9999984904867634
64
-
},
65
-
{
66
-
"class": "yes_male_underwear",
67
-
"score": 1.5095132367094679e-06
68
-
},
69
-
{
70
-
"class": "no_sex_toy",
71
-
"score": 0.9999970970762551
72
-
},
73
-
{
74
-
"class": "yes_sex_toy",
75
-
"score": 2.9029237450490604e-06
76
-
},
77
-
{
78
-
"class": "no_female_nudity",
79
-
"score": 0.9999739028909301
80
-
},
81
-
{
82
-
"class": "yes_female_nudity",
83
-
"score": 2.60971090699536e-05
84
-
},
85
-
{
86
-
"class": "no_male_nudity",
87
-
"score": 0.9999711373083747
88
-
},
89
-
{
90
-
"class": "yes_male_nudity",
91
-
"score": 2.8862691625255323e-05
92
-
},
93
-
{
94
-
"class": "no_female_swimwear",
95
-
"score": 0.9999917609899659
96
-
},
97
-
{
98
-
"class": "yes_female_swimwear",
99
-
"score": 8.239010034025379e-06
100
-
},
101
-
{
102
-
"class": "no_male_shirtless",
103
-
"score": 0.9999583350744331
104
-
},
105
-
{
106
-
"class": "yes_male_shirtless",
107
-
"score": 4.166492556688088e-05
108
-
},
109
-
{
110
-
"class": "no_text",
111
-
"score": 0.9958378716447616
112
-
},
113
-
{
114
-
"class": "text",
115
-
"score": 0.0041621283552384265
116
-
},
117
-
{
118
-
"class": "animated",
119
-
"score": 0.46755478950048235
120
-
},
121
-
{
122
-
"class": "hybrid",
123
-
"score": 0.0011440363434524984
124
-
},
125
-
{
126
-
"class": "natural",
127
-
"score": 0.5313011741560651
128
-
},
129
-
{
130
-
"class": "animated_gun",
131
-
"score": 2.0713000782979496e-05
132
-
},
133
-
{
134
-
"class": "gun_in_hand",
135
-
"score": 1.5844730446534659e-06
136
-
},
137
-
{
138
-
"class": "gun_not_in_hand",
139
-
"score": 1.0338973818006654e-06
140
-
},
141
-
{
142
-
"class": "no_gun",
143
-
"score": 0.9999766686287906
144
-
},
145
-
{
146
-
"class": "culinary_knife_in_hand",
147
-
"score": 3.8063500083369785e-06
148
-
},
149
-
{
150
-
"class": "culinary_knife_not_in_hand",
151
-
"score": 7.94057948996249e-07
152
-
},
153
-
{
154
-
"class": "knife_in_hand",
155
-
"score": 4.5578955723278505e-07
156
-
},
157
-
{
158
-
"class": "knife_not_in_hand",
159
-
"score": 3.842124714748908e-07
160
-
},
161
-
{
162
-
"class": "no_knife",
163
-
"score": 0.999994559590014
164
-
},
165
-
{
166
-
"class": "a_little_bloody",
167
-
"score": 2.1317745626539786e-07
168
-
},
169
-
{
170
-
"class": "no_blood",
171
-
"score": 0.9999793341236429
172
-
},
173
-
{
174
-
"class": "other_blood",
175
-
"score": 2.0322054269591763e-05
176
-
},
177
-
{
178
-
"class": "very_bloody",
179
-
"score": 1.306446309561673e-07
180
-
},
181
-
{
182
-
"class": "no_pills",
183
-
"score": 0.9999989592376954
184
-
},
185
-
{
186
-
"class": "yes_pills",
187
-
"score": 1.0407623044588633e-06
188
-
},
189
-
{
190
-
"class": "no_smoking",
191
-
"score": 0.9999939101969173
192
-
},
193
-
{
194
-
"class": "yes_smoking",
195
-
"score": 6.089803082758281e-06
196
-
},
197
-
{
198
-
"class": "illicit_injectables",
199
-
"score": 6.925695592003094e-07
200
-
},
201
-
{
202
-
"class": "medical_injectables",
203
-
"score": 8.587808234452378e-07
204
-
},
205
-
{
206
-
"class": "no_injectables",
207
-
"score": 0.9999984486496174
208
-
},
209
-
{
210
-
"class": "no_nazi",
211
-
"score": 0.9999987449628097
212
-
},
213
-
{
214
-
"class": "yes_nazi",
215
-
"score": 1.2550371902234279e-06
216
-
},
217
-
{
218
-
"class": "no_kkk",
219
-
"score": 0.999999762417549
220
-
},
221
-
{
222
-
"class": "yes_kkk",
223
-
"score": 2.3758245111050425e-07
224
-
},
225
-
{
226
-
"class": "no_middle_finger",
227
-
"score": 0.9999881515231847
228
-
},
229
-
{
230
-
"class": "yes_middle_finger",
231
-
"score": 1.184847681536747e-05
232
-
},
233
-
{
234
-
"class": "no_terrorist",
235
-
"score": 0.9999998870793229
236
-
},
237
-
{
238
-
"class": "yes_terrorist",
239
-
"score": 1.1292067715380635e-07
240
-
},
241
-
{
242
-
"class": "no_overlay_text",
243
-
"score": 0.9996453363440359
244
-
},
245
-
{
246
-
"class": "yes_overlay_text",
247
-
"score": 0.0003546636559640924
248
-
},
249
-
{
250
-
"class": "no_sexual_activity",
251
-
"score": 0.9999563580374798
252
-
},
253
-
{
254
-
"class": "yes_sexual_activity",
255
-
"score": 0.99,
256
-
"realScore": 4.364196252012032e-05
257
-
},
258
-
{
259
-
"class": "hanging",
260
-
"score": 3.6435135762510905e-07
261
-
},
262
-
{
263
-
"class": "no_hanging_no_noose",
264
-
"score": 0.9999980779196416
265
-
},
266
-
{
267
-
"class": "noose",
268
-
"score": 1.5577290007796094e-06
269
-
},
270
-
{
271
-
"class": "no_realistic_nsfw",
272
-
"score": 0.9999944341007805
273
-
},
274
-
{
275
-
"class": "yes_realistic_nsfw",
276
-
"score": 5.565899219571182e-06
277
-
},
278
-
{
279
-
"class": "animated_corpse",
280
-
"score": 5.276802046755426e-07
281
-
},
282
-
{
283
-
"class": "human_corpse",
284
-
"score": 2.5449360984211012e-08
285
-
},
286
-
{
287
-
"class": "no_corpse",
288
-
"score": 0.9999994468704343
289
-
},
290
-
{
291
-
"class": "no_self_harm",
292
-
"score": 0.9999994515625507
293
-
},
294
-
{
295
-
"class": "yes_self_harm",
296
-
"score": 5.484374493605692e-07
297
-
},
298
-
{
299
-
"class": "no_drawing",
300
-
"score": 0.9978276028816608
301
-
},
302
-
{
303
-
"class": "yes_drawing",
304
-
"score": 0.0021723971183392485
305
-
},
306
-
{
307
-
"class": "no_emaciated_body",
308
-
"score": 0.9999998146500432
309
-
},
310
-
{
311
-
"class": "yes_emaciated_body",
312
-
"score": 1.853499568724518e-07
313
-
},
314
-
{
315
-
"class": "no_child_present",
316
-
"score": 0.9999970498515446
317
-
},
318
-
{
319
-
"class": "yes_child_present",
320
-
"score": 2.950148455380443e-06
321
-
},
322
-
{
323
-
"class": "no_sexual_intent",
324
-
"score": 0.9999963861546292
325
-
},
326
-
{
327
-
"class": "yes_sexual_intent",
328
-
"score": 3.613845370766111e-06
329
-
},
330
-
{
331
-
"class": "animal_genitalia_and_human",
332
-
"score": 2.255472023465222e-08
333
-
},
334
-
{
335
-
"class": "animal_genitalia_only",
336
-
"score": 4.6783185199931176e-07
337
-
},
338
-
{
339
-
"class": "animated_animal_genitalia",
340
-
"score": 6.707857419436447e-07
341
-
},
342
-
{
343
-
"class": "no_animal_genitalia",
344
-
"score": 0.9999988388276858
345
-
},
346
-
{
347
-
"class": "no_gambling",
348
-
"score": 0.9999960939687145
349
-
},
350
-
{
351
-
"class": "yes_gambling",
352
-
"score": 3.906031285604864e-06
353
-
},
354
-
{
355
-
"class": "no_undressed",
356
-
"score": 0.99999923356218
357
-
},
358
-
{
359
-
"class": "yes_undressed",
360
-
"score": 7.664378199789045e-07
361
-
},
362
-
{
363
-
"class": "no_confederate",
364
-
"score": 0.9999925456900376
365
-
},
366
-
{
367
-
"class": "yes_confederate",
368
-
"score": 7.454309962453175e-06
369
-
},
370
-
{
371
-
"class": "animated_alcohol",
372
-
"score": 1.8109949948066074e-06
373
-
},
374
-
{
375
-
"class": "no_alcohol",
376
-
"score": 0.9999916620957963
377
-
},
378
-
{
379
-
"class": "yes_alcohol",
380
-
"score": 5.88781463445443e-06
381
-
},
382
-
{
383
-
"class": "yes_drinking_alcohol",
384
-
"score": 6.390945746578106e-07
385
-
},
386
-
{
387
-
"class": "no_religious_icon",
388
-
"score": 0.9999862158580689
389
-
},
390
-
{
391
-
"class": "yes_religious_icon",
392
-
"score": 1.3784141931119298e-05
393
-
}
394
-
]
395
-
}
396
-
]
397
-
}
398
-
}
399
-
],
400
-
"from_cache": false
401
-
}
-107
labeler/util.go
-107
labeler/util.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"crypto/ecdsa"
5
-
"crypto/elliptic"
6
-
"crypto/rand"
7
-
"encoding/json"
8
-
"errors"
9
-
"fmt"
10
-
"os"
11
-
"path/filepath"
12
-
13
-
"github.com/lestrrat-go/jwx/v2/jwa"
14
-
"github.com/lestrrat-go/jwx/v2/jwk"
15
-
"github.com/whyrusleeping/go-did"
16
-
)
17
-
18
-
// TODO:(bnewbold): duplicates elsewhere; should refactor into cliutil
19
-
func LoadOrCreateKeyFile(kfile, kid string) (*did.PrivKey, error) {
20
-
_, err := os.Stat(kfile)
21
-
if errors.Is(err, os.ErrNotExist) {
22
-
// file doesn't exist; create a new key and write it out, then we will re-read it
23
-
err = CreateKeyFile(kfile, kid)
24
-
if err != nil {
25
-
return nil, err
26
-
}
27
-
}
28
-
29
-
kb, err := os.ReadFile(kfile)
30
-
if err != nil {
31
-
return nil, err
32
-
}
33
-
34
-
return ParseSecretKey(string(kb))
35
-
}
36
-
37
-
func CreateKeyFile(kfile, kid string) error {
38
-
39
-
raw, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
40
-
if err != nil {
41
-
return fmt.Errorf("failed to generate new ECDSA private key: %s", err)
42
-
}
43
-
44
-
key, err := jwk.FromRaw(raw)
45
-
if err != nil {
46
-
return fmt.Errorf("failed to create ECDSA key: %s", err)
47
-
}
48
-
49
-
if _, ok := key.(jwk.ECDSAPrivateKey); !ok {
50
-
return fmt.Errorf("expected jwk.ECDSAPrivateKey, got %T", key)
51
-
}
52
-
53
-
key.Set(jwk.KeyIDKey, kid)
54
-
55
-
buf, err := json.MarshalIndent(key, "", " ")
56
-
if err != nil {
57
-
return fmt.Errorf("failed to marshal key into JSON: %w", err)
58
-
}
59
-
60
-
// ensure data directory exists; won't error if it does
61
-
os.MkdirAll(filepath.Dir(kfile), os.ModePerm)
62
-
63
-
return os.WriteFile(kfile, buf, 0664)
64
-
}
65
-
66
-
func ParseSecretKey(val string) (*did.PrivKey, error) {
67
-
68
-
sk, err := jwk.ParseKey([]byte(val))
69
-
if err != nil {
70
-
return nil, err
71
-
}
72
-
73
-
var spk ecdsa.PrivateKey
74
-
if err := sk.Raw(&spk); err != nil {
75
-
return nil, err
76
-
}
77
-
curve, ok := sk.Get("crv")
78
-
if !ok {
79
-
return nil, fmt.Errorf("need a curve set")
80
-
}
81
-
82
-
var out string
83
-
kts := string(curve.(jwa.EllipticCurveAlgorithm))
84
-
switch kts {
85
-
case "P-256":
86
-
out = did.KeyTypeP256
87
-
default:
88
-
return nil, fmt.Errorf("unrecognized key type: %s", kts)
89
-
}
90
-
91
-
return &did.PrivKey{
92
-
Raw: &spk,
93
-
Type: out,
94
-
}, nil
95
-
}
96
-
97
-
func dedupeStrings(in []string) []string {
98
-
var out []string
99
-
seen := make(map[string]bool)
100
-
for _, v := range in {
101
-
if !seen[v] {
102
-
out = append(out, v)
103
-
seen[v] = true
104
-
}
105
-
}
106
-
return out
107
-
}
-27
labeler/util_test.go
-27
labeler/util_test.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"fmt"
5
-
"reflect"
6
-
"testing"
7
-
)
8
-
9
-
func TestDedupeStrings(t *testing.T) {
10
-
11
-
testCases := []struct {
12
-
in []string
13
-
out []string
14
-
}{
15
-
{in: []string{"a", "b", "c"}, out: []string{"a", "b", "c"}},
16
-
{in: []string{"a", "a", "a"}, out: []string{"a"}},
17
-
{in: []string{"a", "b", "a"}, out: []string{"a", "b"}},
18
-
}
19
-
20
-
for _, c := range testCases {
21
-
res := dedupeStrings(c.in)
22
-
if !reflect.DeepEqual(res, c.out) {
23
-
t.Log(fmt.Sprintf("strings expected:%s got:%s", c.out, res))
24
-
t.Fail()
25
-
}
26
-
}
27
-
}
-81
labeler/ws_endpoints.go
-81
labeler/ws_endpoints.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"fmt"
5
-
"strconv"
6
-
7
-
"github.com/bluesky-social/indigo/events"
8
-
lexutil "github.com/bluesky-social/indigo/lex/util"
9
-
10
-
"github.com/gorilla/websocket"
11
-
"github.com/labstack/echo/v4"
12
-
)
13
-
14
-
func (s *Server) EventsLabelsWebsocket(c echo.Context) error {
15
-
var since *int64
16
-
if sinceVal := c.QueryParam("cursor"); sinceVal != "" {
17
-
sval, err := strconv.ParseInt(sinceVal, 10, 64)
18
-
if err != nil {
19
-
return err
20
-
}
21
-
since = &sval
22
-
}
23
-
24
-
ctx := c.Request().Context()
25
-
26
-
conn, err := websocket.Upgrade(c.Response().Writer, c.Request(), c.Response().Header(), 1<<10, 1<<10)
27
-
if err != nil {
28
-
return fmt.Errorf("upgrading websocket: %w", err)
29
-
}
30
-
31
-
ident := c.RealIP() + "-" + c.Request().UserAgent()
32
-
33
-
evts, cancel, err := s.evtmgr.Subscribe(ctx, ident, func(evt *events.XRPCStreamEvent) bool {
34
-
return true
35
-
}, since)
36
-
if err != nil {
37
-
return err
38
-
}
39
-
defer cancel()
40
-
41
-
header := events.EventHeader{Op: events.EvtKindMessage}
42
-
for {
43
-
select {
44
-
case evt := <-evts:
45
-
wc, err := conn.NextWriter(websocket.BinaryMessage)
46
-
if err != nil {
47
-
return err
48
-
}
49
-
50
-
var obj lexutil.CBOR
51
-
52
-
switch {
53
-
case evt.Error != nil:
54
-
header.Op = events.EvtKindErrorFrame
55
-
obj = evt.Error
56
-
case evt.LabelInfo != nil:
57
-
header.MsgType = "#info"
58
-
obj = evt.LabelInfo
59
-
case evt.LabelLabels != nil:
60
-
header.MsgType = "#labels"
61
-
obj = evt.LabelLabels
62
-
default:
63
-
return fmt.Errorf("unrecognized event kind")
64
-
}
65
-
66
-
if err := header.MarshalCBOR(wc); err != nil {
67
-
return fmt.Errorf("failed to write header: %w", err)
68
-
}
69
-
70
-
if err := obj.MarshalCBOR(wc); err != nil {
71
-
return fmt.Errorf("failed to write event: %w", err)
72
-
}
73
-
74
-
if err := wc.Close(); err != nil {
75
-
return fmt.Errorf("failed to flush-close our event write: %w", err)
76
-
}
77
-
case <-ctx.Done():
78
-
return nil
79
-
}
80
-
}
81
-
}
-137
labeler/xrpc_endpoints.go
-137
labeler/xrpc_endpoints.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"net/http/httputil"
5
-
"strconv"
6
-
7
-
atproto "github.com/bluesky-social/indigo/api/atproto"
8
-
9
-
"github.com/carlmjohnson/versioninfo"
10
-
"github.com/labstack/echo/v4"
11
-
"go.opentelemetry.io/otel"
12
-
)
13
-
14
-
func (s *Server) RegisterHandlersComAtproto(e *echo.Echo) error {
15
-
// handle/account hosting
16
-
e.GET("/xrpc/com.atproto.server.describeServer", s.HandleComAtprotoServerDescribeServer)
17
-
// TODO: session create/refresh/delete?
18
-
19
-
// minimal moderation reporting/actioning
20
-
e.POST("/xrpc/com.atproto.report.create", s.HandleComAtprotoReportCreate)
21
-
22
-
// label-specific
23
-
e.GET("/xrpc/com.atproto.label.queryLabels", s.HandleComAtprotoLabelQueryLabels)
24
-
25
-
return nil
26
-
}
27
-
28
-
func (s *Server) rewriteProxyRequestAdmin(r *httputil.ProxyRequest) {
29
-
r.SetXForwarded()
30
-
r.SetURL(s.xrpcProxyURL)
31
-
r.Out.Header.Set("Authorization", s.xrpcProxyAuthHeader)
32
-
}
33
-
34
-
func (s *Server) RegisterProxyHandlers(e *echo.Echo) error {
35
-
36
-
rp := &httputil.ReverseProxy{Rewrite: s.rewriteProxyRequestAdmin}
37
-
38
-
// Proxy some admin requests
39
-
e.GET("/xrpc/com.atproto.admin.getRecord", echo.WrapHandler(rp))
40
-
e.GET("/xrpc/com.atproto.admin.getRepo", echo.WrapHandler(rp))
41
-
e.GET("/xrpc/com.atproto.admin.searchRepos", echo.WrapHandler(rp))
42
-
43
-
return nil
44
-
}
45
-
46
-
func (s *Server) HandleComAtprotoServerDescribeServer(c echo.Context) error {
47
-
ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandleComAtprotoServerDescribeServer")
48
-
defer span.End()
49
-
var out *atproto.ServerDescribeServer_Output
50
-
var handleErr error
51
-
// func (s *Server) handleComAtprotoServerDescribeServer(ctx context.Context) (*atproto.ServerDescribeServer_Output, error)
52
-
out, handleErr = s.handleComAtprotoServerDescribeServer(ctx)
53
-
if handleErr != nil {
54
-
return handleErr
55
-
}
56
-
return c.JSON(200, out)
57
-
}
58
-
59
-
func (s *Server) HandleComAtprotoLabelQueryLabels(c echo.Context) error {
60
-
ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandleComAtprotoLabelQueryLabels")
61
-
defer span.End()
62
-
cursor := c.QueryParam("cursor")
63
-
64
-
var limit int
65
-
if p := c.QueryParam("limit"); p != "" {
66
-
var err error
67
-
limit, err = strconv.Atoi(p)
68
-
if err != nil {
69
-
return err
70
-
}
71
-
} else {
72
-
limit = 20
73
-
}
74
-
75
-
sources := c.QueryParams()["sources"]
76
-
77
-
uriPatterns := c.QueryParams()["uriPatterns"]
78
-
var out *atproto.LabelQueryLabels_Output
79
-
var handleErr error
80
-
// func (s *Server) handleComAtprotoLabelQueryLabels(ctx context.Context,cursor string,limit int,sources []string,uriPatterns []string) (*comatprototypes.LabelQueryLabels_Output, error)
81
-
out, handleErr = s.handleComAtprotoLabelQueryLabels(ctx, cursor, limit, sources, uriPatterns)
82
-
if handleErr != nil {
83
-
return handleErr
84
-
}
85
-
return c.JSON(200, out)
86
-
}
87
-
88
-
func (s *Server) HandleComAtprotoAdminGetModerationReport(c echo.Context) error {
89
-
ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandleComAtprotoAdminGetModerationReport")
90
-
defer span.End()
91
-
92
-
id, err := strconv.Atoi(c.QueryParam("id"))
93
-
if err != nil {
94
-
return err
95
-
}
96
-
var out *atproto.AdminDefs_ReportViewDetail
97
-
var handleErr error
98
-
// func (s *Server) handleComAtprotoAdminGetModerationReport(ctx context.Context,id int) (*atproto.AdminDefs_ReportViewDetail, error)
99
-
out, handleErr = s.handleComAtprotoAdminGetModerationReport(ctx, id)
100
-
if handleErr != nil {
101
-
return handleErr
102
-
}
103
-
return c.JSON(200, out)
104
-
}
105
-
106
-
func (s *Server) HandleComAtprotoReportCreate(c echo.Context) error {
107
-
ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandleComAtprotoReportCreate")
108
-
defer span.End()
109
-
110
-
var body atproto.ModerationCreateReport_Input
111
-
if err := c.Bind(&body); err != nil {
112
-
return err
113
-
}
114
-
var out *atproto.ModerationCreateReport_Output
115
-
var handleErr error
116
-
// func (s *Server) handleComAtprotoReportCreate(ctx context.Context,body *atproto.ModerationCreateReport_Input) (*atproto.ModerationCreateReport_Output, error)
117
-
out, handleErr = s.handleComAtprotoReportCreate(ctx, &body)
118
-
if handleErr != nil {
119
-
return handleErr
120
-
}
121
-
return c.JSON(200, out)
122
-
}
123
-
124
-
type HealthStatus struct {
125
-
Status string `json:"status"`
126
-
Version string `json:"version"`
127
-
Message string `json:"msg,omitempty"`
128
-
}
129
-
130
-
func (s *Server) HandleHealthCheck(c echo.Context) error {
131
-
if err := s.db.Exec("SELECT 1").Error; err != nil {
132
-
log.Errorf("healthcheck can't connect to database: %v", err)
133
-
return c.JSON(500, HealthStatus{Status: "error", Version: versioninfo.Short(), Message: "can't connect to database"})
134
-
} else {
135
-
return c.JSON(200, HealthStatus{Status: "ok", Version: versioninfo.Short()})
136
-
}
137
-
}
-192
labeler/xrpc_handlers.go
-192
labeler/xrpc_handlers.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"context"
5
-
"strconv"
6
-
"strings"
7
-
"time"
8
-
9
-
atproto "github.com/bluesky-social/indigo/api/atproto"
10
-
"github.com/bluesky-social/indigo/models"
11
-
"github.com/bluesky-social/indigo/util"
12
-
13
-
"github.com/labstack/echo/v4"
14
-
)
15
-
16
-
func (s *Server) handleComAtprotoServerDescribeServer(ctx context.Context) (*atproto.ServerDescribeServer_Output, error) {
17
-
invcode := true
18
-
return &atproto.ServerDescribeServer_Output{
19
-
InviteCodeRequired: &invcode,
20
-
AvailableUserDomains: []string{},
21
-
Links: &atproto.ServerDescribeServer_Links{},
22
-
}, nil
23
-
}
24
-
25
-
func (s *Server) handleComAtprotoLabelQueryLabels(ctx context.Context, cursor string, limit int, sources, uriPatterns []string) (*atproto.LabelQueryLabels_Output, error) {
26
-
27
-
if limit <= 0 {
28
-
limit = 20
29
-
}
30
-
if limit > 100 {
31
-
limit = 100
32
-
}
33
-
34
-
q := s.db.Limit(limit).Order("id desc")
35
-
if cursor != "" {
36
-
cursorID, err := strconv.Atoi(cursor)
37
-
if err != nil {
38
-
return nil, err
39
-
}
40
-
q = q.Where("id < ?", cursorID)
41
-
}
42
-
43
-
srcQuery := s.db
44
-
for _, src := range sources {
45
-
if src == "*" {
46
-
continue
47
-
}
48
-
srcQuery = srcQuery.Or("source_did = ?", src)
49
-
}
50
-
if srcQuery != s.db {
51
-
q = q.Where(srcQuery)
52
-
}
53
-
54
-
uriQuery := s.db
55
-
for _, pat := range uriPatterns {
56
-
if strings.HasSuffix(pat, "*") {
57
-
likePat := []rune(pat)
58
-
likePat[len(likePat)-1] = '%'
59
-
uriQuery = uriQuery.Or("uri LIKE ?", string(likePat))
60
-
} else {
61
-
uriQuery = uriQuery.Or("uri = ?", pat)
62
-
}
63
-
}
64
-
if uriQuery != s.db {
65
-
q = q.Where(uriQuery)
66
-
}
67
-
68
-
var labelRows []models.Label
69
-
result := q.Find(&labelRows)
70
-
if result.Error != nil {
71
-
return nil, result.Error
72
-
}
73
-
74
-
var nextCursor string
75
-
if len(labelRows) >= 1 && len(labelRows) == limit {
76
-
nextCursor = strconv.FormatUint(labelRows[len(labelRows)-1].ID, 10)
77
-
}
78
-
79
-
labelObjs := []*atproto.LabelDefs_Label{}
80
-
for _, row := range labelRows {
81
-
neg := false
82
-
if row.Neg != nil && *row.Neg == true {
83
-
neg = true
84
-
}
85
-
labelObjs = append(labelObjs, &atproto.LabelDefs_Label{
86
-
Src: row.SourceDid,
87
-
Uri: row.Uri,
88
-
Cid: row.Cid,
89
-
Val: row.Val,
90
-
Neg: &neg,
91
-
Cts: row.CreatedAt.Format(util.ISO8601),
92
-
})
93
-
}
94
-
out := atproto.LabelQueryLabels_Output{
95
-
Labels: labelObjs,
96
-
}
97
-
if nextCursor != "" {
98
-
out.Cursor = &nextCursor
99
-
}
100
-
return &out, nil
101
-
}
102
-
103
-
func (s *Server) handleComAtprotoAdminGetModerationReport(ctx context.Context, id int) (*atproto.AdminDefs_ReportViewDetail, error) {
104
-
105
-
var row models.ModerationReport
106
-
result := s.db.First(&row, id)
107
-
if result.Error != nil {
108
-
return nil, result.Error
109
-
}
110
-
111
-
full, err := s.hydrateModerationReportDetails(ctx, []models.ModerationReport{row})
112
-
if err != nil {
113
-
return nil, err
114
-
}
115
-
return full[0], nil
116
-
}
117
-
118
-
func didFromURI(uri string) string {
119
-
parts := strings.SplitN(uri, "/", 4)
120
-
if len(parts) < 3 {
121
-
return ""
122
-
}
123
-
if strings.HasPrefix(parts[2], "did:") {
124
-
return parts[2]
125
-
}
126
-
return ""
127
-
}
128
-
129
-
func (s *Server) handleComAtprotoReportCreate(ctx context.Context, body *atproto.ModerationCreateReport_Input) (*atproto.ModerationCreateReport_Output, error) {
130
-
131
-
if body.ReasonType == nil || *body.ReasonType == "" {
132
-
return nil, echo.NewHTTPError(400, "reasonType is required")
133
-
}
134
-
if body.Subject == nil {
135
-
return nil, echo.NewHTTPError(400, "Subject is required")
136
-
}
137
-
138
-
row := models.ModerationReport{
139
-
ReasonType: *body.ReasonType,
140
-
Reason: body.Reason,
141
-
// TODO(bnewbold): temporarily, all reports from labelmaker user
142
-
ReportedByDid: s.user.Did,
143
-
}
144
-
var outSubj atproto.ModerationCreateReport_Output_Subject
145
-
if body.Subject.AdminDefs_RepoRef != nil {
146
-
if body.Subject.AdminDefs_RepoRef.Did == "" {
147
-
return nil, echo.NewHTTPError(400, "DID is required for repo reports")
148
-
}
149
-
row.SubjectType = "com.atproto.repo.repoRef"
150
-
row.SubjectDid = body.Subject.AdminDefs_RepoRef.Did
151
-
outSubj.AdminDefs_RepoRef = &atproto.AdminDefs_RepoRef{
152
-
LexiconTypeID: "com.atproto.repo.repoRef",
153
-
Did: row.SubjectDid,
154
-
}
155
-
} else if body.Subject.RepoStrongRef != nil {
156
-
if body.Subject.RepoStrongRef.Uri == "" {
157
-
return nil, echo.NewHTTPError(400, "URI required for record reports")
158
-
}
159
-
if body.Subject.RepoStrongRef.Cid == "" {
160
-
return nil, echo.NewHTTPError(400, "this implementation requires a strong record ref (aka, with CID) in reports")
161
-
}
162
-
row.SubjectType = "com.atproto.repo.recordRef"
163
-
row.SubjectUri = &body.Subject.RepoStrongRef.Uri
164
-
row.SubjectDid = didFromURI(body.Subject.RepoStrongRef.Uri)
165
-
if row.SubjectDid == "" {
166
-
return nil, echo.NewHTTPError(400, "expected URI with a DID: ", row.SubjectUri)
167
-
}
168
-
row.SubjectCid = &body.Subject.RepoStrongRef.Cid
169
-
outSubj.RepoStrongRef = &atproto.RepoStrongRef{
170
-
LexiconTypeID: "com.atproto.repo.strongRef",
171
-
Uri: *row.SubjectUri,
172
-
Cid: *row.SubjectCid,
173
-
}
174
-
} else {
175
-
return nil, echo.NewHTTPError(400, "report subject must be a repoRef or a recordRef")
176
-
}
177
-
178
-
result := s.db.Create(&row)
179
-
if result.Error != nil {
180
-
return nil, result.Error
181
-
}
182
-
183
-
out := atproto.ModerationCreateReport_Output{
184
-
Id: int64(row.ID),
185
-
CreatedAt: row.CreatedAt.Format(time.RFC3339),
186
-
Reason: row.Reason,
187
-
ReasonType: &row.ReasonType,
188
-
ReportedBy: row.ReportedByDid,
189
-
Subject: &outSubj,
190
-
}
191
-
return &out, nil
192
-
}
-211
labeler/xrpc_test.go
-211
labeler/xrpc_test.go
···
1
-
package labeler
2
-
3
-
import (
4
-
"context"
5
-
"encoding/json"
6
-
"net/http"
7
-
"net/http/httptest"
8
-
"net/url"
9
-
"strings"
10
-
"testing"
11
-
12
-
comatproto "github.com/bluesky-social/indigo/api/atproto"
13
-
14
-
"github.com/labstack/echo/v4"
15
-
"github.com/stretchr/testify/assert"
16
-
)
17
-
18
-
func TestLabelMakerXRPCReportRepo(t *testing.T) {
19
-
assert := assert.New(t)
20
-
e := echo.New()
21
-
lm := testLabelMaker(t)
22
-
23
-
// create and read back a basic repo report
24
-
rt := "spam"
25
-
reportedDid := "did:plc:123"
26
-
report := comatproto.ModerationCreateReport_Input{
27
-
//Reason
28
-
ReasonType: &rt,
29
-
Subject: &comatproto.ModerationCreateReport_Input_Subject{
30
-
AdminDefs_RepoRef: &comatproto.AdminDefs_RepoRef{
31
-
Did: reportedDid,
32
-
},
33
-
},
34
-
}
35
-
out := testCreateReport(t, e, lm, &report)
36
-
assert.Equal(&rt, out.ReasonType)
37
-
assert.Nil(out.Reason)
38
-
assert.Equal(reportedDid, out.Subject.AdminDefs_RepoRef.Did)
39
-
}
40
-
41
-
func TestLabelMakerXRPCReportRepoBad(t *testing.T) {
42
-
assert := assert.New(t)
43
-
e := echo.New()
44
-
lm := testLabelMaker(t)
45
-
46
-
table := []struct {
47
-
rType string
48
-
rDid string
49
-
statusCode int
50
-
}{
51
-
{"spam", "did:plc:123", 200},
52
-
{"", "did:plc:123", 400},
53
-
{"spam", "", 400},
54
-
}
55
-
56
-
for _, row := range table {
57
-
58
-
report := comatproto.ModerationCreateReport_Input{
59
-
//Reason
60
-
ReasonType: &row.rType,
61
-
Subject: &comatproto.ModerationCreateReport_Input_Subject{
62
-
AdminDefs_RepoRef: &comatproto.AdminDefs_RepoRef{
63
-
Did: row.rDid,
64
-
},
65
-
},
66
-
}
67
-
reportJSON, err := json.Marshal(report)
68
-
if err != nil {
69
-
t.Fatal(err)
70
-
}
71
-
req := httptest.NewRequest(http.MethodPost, "/xrpc/com.atproto.report.create", strings.NewReader(string(reportJSON)))
72
-
req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
73
-
recorder := httptest.NewRecorder()
74
-
c := e.NewContext(req, recorder)
75
-
err = lm.HandleComAtprotoReportCreate(c)
76
-
if err != nil {
77
-
httpError, _ := err.(*echo.HTTPError)
78
-
assert.Equal(row.statusCode, httpError.Code)
79
-
} else {
80
-
assert.Equal(row.statusCode, recorder.Code)
81
-
}
82
-
}
83
-
}
84
-
85
-
func TestLabelMakerXRPCReportRecord(t *testing.T) {
86
-
assert := assert.New(t)
87
-
e := echo.New()
88
-
lm := testLabelMaker(t)
89
-
// create a second report, on a record
90
-
rt := "spam"
91
-
reason := "I just don't like it!"
92
-
uri := "at://did:plc:123/com.example.record/bcd234"
93
-
cid := "bafyreie5cvv4h45feadgeuwhbcutmh6t2ceseocckahdoe6uat64zmz454"
94
-
report := comatproto.ModerationCreateReport_Input{
95
-
Reason: &reason,
96
-
ReasonType: &rt,
97
-
Subject: &comatproto.ModerationCreateReport_Input_Subject{
98
-
RepoStrongRef: &comatproto.RepoStrongRef{
99
-
//com.atproto.repo.strongRef
100
-
Uri: uri,
101
-
Cid: cid,
102
-
},
103
-
},
104
-
}
105
-
out := testCreateReport(t, e, lm, &report)
106
-
assert.Equal(report.ReasonType, out.ReasonType)
107
-
assert.Equal(report.Reason, out.Reason)
108
-
assert.NotNil(out.CreatedAt)
109
-
assert.NotNil(out.ReportedBy)
110
-
assert.Equal(report.Subject.AdminDefs_RepoRef, out.Subject.AdminDefs_RepoRef)
111
-
assert.Equal(report.Subject.RepoStrongRef, out.Subject.RepoStrongRef)
112
-
}
113
-
114
-
func TestLabelMakerXRPCReportRecordBad(t *testing.T) {
115
-
assert := assert.New(t)
116
-
e := echo.New()
117
-
lm := testLabelMaker(t)
118
-
119
-
uriStr := "at://did:plc:123/com.example.record/bcd234"
120
-
cidStr := "bafyreie5cvv4h45feadgeuwhbcutmh6t2ceseocckahdoe6uat64zmz454"
121
-
emptyStr := ""
122
-
table := []struct {
123
-
rType string
124
-
rUri string
125
-
rCid string
126
-
statusCode int
127
-
}{
128
-
{"spam", uriStr, cidStr, 200},
129
-
{"", uriStr, cidStr, 400},
130
-
{"spam", "", cidStr, 400},
131
-
{"spam", uriStr, emptyStr, 400},
132
-
}
133
-
134
-
for _, row := range table {
135
-
136
-
report := comatproto.ModerationCreateReport_Input{
137
-
ReasonType: &row.rType,
138
-
Subject: &comatproto.ModerationCreateReport_Input_Subject{
139
-
RepoStrongRef: &comatproto.RepoStrongRef{
140
-
//com.atproto.repo.strongRef
141
-
Uri: row.rUri,
142
-
Cid: row.rCid,
143
-
},
144
-
},
145
-
}
146
-
reportJSON, err := json.Marshal(report)
147
-
if err != nil {
148
-
t.Fatal(err)
149
-
}
150
-
req := httptest.NewRequest(http.MethodPost, "/xrpc/com.atproto.report.create", strings.NewReader(string(reportJSON)))
151
-
req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
152
-
recorder := httptest.NewRecorder()
153
-
c := e.NewContext(req, recorder)
154
-
err = lm.HandleComAtprotoReportCreate(c)
155
-
if err != nil {
156
-
httpError, _ := err.(*echo.HTTPError)
157
-
assert.Equal(row.statusCode, httpError.Code)
158
-
} else {
159
-
assert.Equal(row.statusCode, recorder.Code)
160
-
}
161
-
}
162
-
}
163
-
164
-
func TestLabelMakerXRPCLabelQuery(t *testing.T) {
165
-
assert := assert.New(t)
166
-
e := echo.New()
167
-
lm := testLabelMaker(t)
168
-
ctx := context.TODO()
169
-
170
-
// simple query, no labels
171
-
p1 := make(url.Values)
172
-
p1.Set("uriPatterns", "*")
173
-
out1, err := testQueryLabels(t, e, lm, &p1)
174
-
assert.NoError(err)
175
-
assert.Equal(0, len(out1.Labels))
176
-
177
-
// create a label, then query
178
-
neg := false
179
-
l3 := comatproto.LabelDefs_Label{
180
-
Uri: "at://did:plc:fake/com.example/abc234",
181
-
Val: "example",
182
-
Cts: "2023-03-15T22:16:18.408Z",
183
-
// TODO: temporary hack, should just be null
184
-
Neg: &neg,
185
-
}
186
-
lm.CommitLabels(ctx, []*comatproto.LabelDefs_Label{&l3}, false)
187
-
p3 := make(url.Values)
188
-
p3.Set("uriPatterns", l3.Uri)
189
-
out3, err := testQueryLabels(t, e, lm, &p3)
190
-
assert.NoError(err)
191
-
assert.Equal(1, len(out3.Labels))
192
-
assert.Equal(&l3, out3.Labels[0])
193
-
}
194
-
195
-
func TestDidFromURI(t *testing.T) {
196
-
assert := assert.New(t)
197
-
cases := []struct {
198
-
input string
199
-
expected string
200
-
}{
201
-
{input: "", expected: ""},
202
-
{input: "at://did:plc:fake/com.example/abc234", expected: "did:plc:fake"},
203
-
{input: "at://example.com/com.example/abc234", expected: ""},
204
-
{input: "at://did:plc:fake", expected: "did:plc:fake"},
205
-
}
206
-
207
-
for _, tc := range cases {
208
-
out := didFromURI(tc.input)
209
-
assert.Equal(tc.expected, out)
210
-
}
211
-
}
+4
-1
search/firehose.go
+4
-1
search/firehose.go
+25
-19
search/indexing.go
+25
-19
search/indexing.go
···
6
6
"encoding/json"
7
7
"fmt"
8
8
"io"
9
-
"regexp"
10
9
"strings"
11
10
12
11
appbsky "github.com/bluesky-social/indigo/api/bsky"
···
18
17
esapi "github.com/opensearch-project/opensearch-go/v2/opensearchapi"
19
18
)
20
19
21
-
func (s *Server) deletePost(ctx context.Context, ident *identity.Identity, rkey string) error {
20
+
func (s *Server) deletePost(ctx context.Context, ident *identity.Identity, recordPath string) error {
22
21
ctx, span := tracer.Start(ctx, "deletePost")
23
22
defer span.End()
24
-
span.SetAttributes(attribute.String("repo", ident.DID.String()), attribute.String("rkey", rkey))
23
+
span.SetAttributes(attribute.String("repo", ident.DID.String()), attribute.String("path", recordPath))
25
24
26
-
log := s.logger.With("repo", ident.DID, "rkey", rkey, "op", "deletePost")
27
-
log.Info("deleting post from index")
25
+
logger := s.logger.With("repo", ident.DID, "path", recordPath, "op", "deletePost")
26
+
27
+
parts := strings.SplitN(recordPath, "/", 3)
28
+
if len(parts) < 2 {
29
+
logger.Warn("skipping post record with malformed path")
30
+
return nil
31
+
}
32
+
rkey, err := syntax.ParseTID(parts[1])
33
+
if err != nil {
34
+
logger.Warn("skipping post record with non-TID rkey")
35
+
return nil
36
+
}
37
+
28
38
docID := fmt.Sprintf("%s_%s", ident.DID.String(), rkey)
39
+
logger.Info("deleting post from index", "docID", docID)
29
40
req := esapi.DeleteRequest{
30
41
Index: s.postIndex,
31
42
DocumentID: docID,
···
42
53
return fmt.Errorf("failed to read indexing response: %w", err)
43
54
}
44
55
if res.IsError() {
45
-
log.Warn("opensearch indexing error", "status_code", res.StatusCode, "response", res, "body", string(body))
56
+
logger.Warn("opensearch indexing error", "status_code", res.StatusCode, "response", res, "body", string(body))
46
57
return fmt.Errorf("indexing error, code=%d", res.StatusCode)
47
58
}
48
59
return nil
49
60
}
50
-
51
-
var tidRegex = regexp.MustCompile(`^[234567abcdefghijklmnopqrstuvwxyz]{13}$`)
52
61
53
62
func (s *Server) indexPost(ctx context.Context, ident *identity.Identity, rec *appbsky.FeedPost, path string, rcid cid.Cid) error {
54
63
ctx, span := tracer.Start(ctx, "indexPost")
···
57
66
58
67
log := s.logger.With("repo", ident.DID, "path", path, "op", "indexPost")
59
68
parts := strings.SplitN(path, "/", 3)
60
-
// TODO: replace with an atproto/syntax package type for TID
61
-
if len(parts) != 2 || !tidRegex.MatchString(parts[1]) {
62
-
log.Warn("skipping index post record with weird path/TID", "did", ident.DID, "path", path)
69
+
if len(parts) < 2 {
70
+
log.Warn("skipping post record with malformed path")
63
71
return nil
64
72
}
65
-
rkey := parts[1]
66
-
67
-
log = log.With("rkey", rkey)
68
-
69
-
_, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
73
+
rkey, err := syntax.ParseTID(parts[1])
70
74
if err != nil {
71
-
log.Warn("post had invalid timestamp", "createdAt", rec.CreatedAt, "parseErr", err)
72
-
rec.CreatedAt = ""
75
+
log.Warn("skipping post record with non-TID rkey")
76
+
return nil
73
77
}
74
78
75
-
doc := TransformPost(rec, ident, rkey, rcid.String())
79
+
log = log.With("rkey", rkey)
80
+
81
+
doc := TransformPost(rec, ident, rkey.String(), rcid.String())
76
82
b, err := json.Marshal(doc)
77
83
if err != nil {
78
84
return err
+14
-8
search/server.go
+14
-8
search/server.go
···
37
37
38
38
bfs *backfill.Gormstore
39
39
bf *backfill.Backfiller
40
+
41
+
enableRepoDiscovery bool
40
42
}
41
43
42
44
type LastSeq struct {
···
51
53
Logger *slog.Logger
52
54
BGSSyncRateLimit int
53
55
IndexMaxConcurrency int
56
+
DiscoverRepos bool
54
57
}
55
58
56
59
func NewServer(db *gorm.DB, escli *es.Client, dir identity.Directory, config Config) (*Server, error) {
···
76
79
}
77
80
78
81
s := &Server{
79
-
escli: escli,
80
-
profileIndex: config.ProfileIndex,
81
-
postIndex: config.PostIndex,
82
-
db: db,
83
-
bgshost: config.BGSHost, // NOTE: the original URL, not 'bgshttp'
84
-
bgsxrpc: bgsxrpc,
85
-
dir: dir,
86
-
logger: logger,
82
+
escli: escli,
83
+
profileIndex: config.ProfileIndex,
84
+
postIndex: config.PostIndex,
85
+
db: db,
86
+
bgshost: config.BGSHost, // NOTE: the original URL, not 'bgshttp'
87
+
bgsxrpc: bgsxrpc,
88
+
dir: dir,
89
+
logger: logger,
90
+
enableRepoDiscovery: config.DiscoverRepos,
87
91
}
88
92
89
93
bfstore := backfill.NewGormstore(db)
···
164
168
}
165
169
166
170
type HealthStatus struct {
171
+
Service string `json:"service,const=palomar"`
167
172
Status string `json:"status"`
168
173
Version string `json:"version"`
169
174
Message string `json:"msg,omitempty"`
···
198
203
}
199
204
200
205
e.Use(middleware.CORS())
206
+
e.GET("/", s.handleHealthCheck)
201
207
e.GET("/_health", s.handleHealthCheck)
202
208
e.GET("/metrics", echo.WrapHandler(promhttp.Handler()))
203
209
e.GET("/xrpc/app.bsky.unspecced.searchPostsSkeleton", s.handleSearchPostsSkeleton)
+10
-5
search/transform.go
+10
-5
search/transform.go
···
2
2
3
3
import (
4
4
"strings"
5
-
"time"
6
5
7
6
appbsky "github.com/bluesky-social/indigo/api/bsky"
8
7
"github.com/bluesky-social/indigo/atproto/identity"
9
-
"github.com/bluesky-social/indigo/util"
8
+
"github.com/bluesky-social/indigo/atproto/syntax"
9
+
10
10
"github.com/rivo/uniseg"
11
11
)
12
12
···
80
80
handle = ident.Handle.String()
81
81
}
82
82
return ProfileDoc{
83
-
DocIndexTs: time.Now().UTC().Format(util.ISO8601),
83
+
DocIndexTs: syntax.DatetimeNow().String(),
84
84
DID: ident.DID.String(),
85
85
RecordCID: cid,
86
86
Handle: handle,
···
157
157
}
158
158
159
159
doc := PostDoc{
160
-
DocIndexTs: time.Now().UTC().Format(util.ISO8601),
160
+
DocIndexTs: syntax.DatetimeNow().String(),
161
161
DID: ident.DID.String(),
162
162
RecordRkey: rkey,
163
163
RecordCID: cid,
···
177
177
}
178
178
179
179
if post.CreatedAt != "" {
180
-
doc.CreatedAt = &post.CreatedAt
180
+
// there are some old bad timestamps out there!
181
+
dt, err := syntax.ParseDatetimeLenient(post.CreatedAt)
182
+
if nil == err { // *not* an error
183
+
s := dt.String()
184
+
doc.CreatedAt = &s
185
+
}
181
186
}
182
187
183
188
return doc
-174
testing/labelmaker_fakedata_test.go
-174
testing/labelmaker_fakedata_test.go
···
1
-
package testing
2
-
3
-
import (
4
-
"context"
5
-
"fmt"
6
-
"net/http"
7
-
"os"
8
-
"path/filepath"
9
-
"testing"
10
-
"time"
11
-
12
-
comatproto "github.com/bluesky-social/indigo/api/atproto"
13
-
"github.com/bluesky-social/indigo/carstore"
14
-
"github.com/bluesky-social/indigo/events"
15
-
"github.com/bluesky-social/indigo/events/schedulers/sequential"
16
-
"github.com/bluesky-social/indigo/labeler"
17
-
"github.com/bluesky-social/indigo/util"
18
-
"github.com/bluesky-social/indigo/xrpc"
19
-
20
-
"github.com/gorilla/websocket"
21
-
"github.com/stretchr/testify/assert"
22
-
"gorm.io/driver/sqlite"
23
-
"gorm.io/gorm"
24
-
)
25
-
26
-
func testLabelMaker(t *testing.T) *labeler.Server {
27
-
28
-
tempdir, err := os.MkdirTemp("", "labelmaker-test-")
29
-
if err != nil {
30
-
t.Fatal(err)
31
-
}
32
-
sharddir := filepath.Join(tempdir, "shards")
33
-
if err := os.MkdirAll(sharddir, 0775); err != nil {
34
-
t.Fatal(err)
35
-
}
36
-
37
-
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{SkipDefaultTransaction: true})
38
-
if err != nil {
39
-
t.Fatal(err)
40
-
}
41
-
42
-
cs, err := carstore.NewCarStore(db, sharddir)
43
-
if err != nil {
44
-
t.Fatal(err)
45
-
}
46
-
47
-
repoKeyPath := filepath.Join(tempdir, "labelmaker.key")
48
-
serkey, err := labeler.LoadOrCreateKeyFile(repoKeyPath, "auto-labelmaker")
49
-
if err != nil {
50
-
t.Fatal(err)
51
-
}
52
-
53
-
plcURL := "http://did-plc-test.dummy"
54
-
blobPdsURL := "http://pds-test.dummy"
55
-
repoUser := labeler.RepoConfig{
56
-
Handle: "test.handle.dummy",
57
-
Did: "did:plc:testdummy",
58
-
Password: "test-admin-pass",
59
-
SigningKey: serkey,
60
-
UserId: 1,
61
-
}
62
-
xrpcProxyURL := "http://proxy-test.dummy"
63
-
xrpcProxyAdminPassword := "test-dummy-password"
64
-
65
-
lm, err := labeler.NewServer(db, cs, repoUser, plcURL, blobPdsURL, xrpcProxyURL, xrpcProxyAdminPassword, false)
66
-
if err != nil {
67
-
t.Fatal(err)
68
-
}
69
-
return lm
70
-
}
71
-
72
-
func labelEvents(t *testing.T, lm *labeler.Server, since int64) *EventStream {
73
-
d := websocket.Dialer{}
74
-
h := http.Header{}
75
-
bgsHost := "localhost:1234"
76
-
77
-
q := ""
78
-
if since >= 0 {
79
-
q = fmt.Sprintf("?cursor=%d", since)
80
-
}
81
-
82
-
con, resp, err := d.Dial("ws://"+bgsHost+"/xrpc/com.atproto.sync.subscribeLabels"+q, h)
83
-
if err != nil {
84
-
t.Fatal(err)
85
-
}
86
-
87
-
if resp.StatusCode != 101 {
88
-
t.Fatal("expected http 101 response, got: ", resp.StatusCode)
89
-
}
90
-
91
-
ctx, cancel := context.WithCancel(context.Background())
92
-
93
-
es := &EventStream{
94
-
Cancel: cancel,
95
-
}
96
-
97
-
go func() {
98
-
<-ctx.Done()
99
-
con.Close()
100
-
}()
101
-
102
-
go func() {
103
-
rsc := &events.RepoStreamCallbacks{
104
-
LabelLabels: func(evt *comatproto.LabelSubscribeLabels_Labels) error {
105
-
fmt.Println("received event: ", evt.Seq)
106
-
es.Lk.Lock()
107
-
es.Events = append(es.Events, &events.XRPCStreamEvent{LabelLabels: evt})
108
-
es.Lk.Unlock()
109
-
return nil
110
-
},
111
-
}
112
-
seqScheduler := sequential.NewScheduler("test", rsc.EventHandler)
113
-
if err := events.HandleRepoStream(ctx, con, seqScheduler); err != nil {
114
-
fmt.Println(err)
115
-
}
116
-
}()
117
-
118
-
return es
119
-
}
120
-
121
-
/*
122
-
labelmaker interop:
123
-
- create golang PDS+BGS+labelmaker
124
-
- create user and posts
125
-
- check labelmaker state
126
-
*/
127
-
128
-
func TestLabelmakerBasic(t *testing.T) {
129
-
assert := assert.New(t)
130
-
_ = assert
131
-
ctx := context.TODO()
132
-
didr := TestPLC(t)
133
-
p1 := MustSetupPDS(t, ".tpds", didr)
134
-
p1.Run(t)
135
-
136
-
b1 := MustSetupBGS(t, didr)
137
-
b1.Run(t)
138
-
139
-
p1.RequestScraping(t, b1)
140
-
p1.BumpLimits(t, b1)
141
-
142
-
l1 := testLabelMaker(t)
143
-
l1.AddKeywordLabeler(labeler.KeywordLabeler{Value: "definite-article", Keywords: []string{"the"}})
144
-
go l1.RunAPI(":7711")
145
-
defer l1.Shutdown(ctx)
146
-
147
-
time.Sleep(time.Millisecond * 50)
148
-
149
-
evts := b1.Events(t, -1)
150
-
defer evts.Cancel()
151
-
152
-
bob := p1.MustNewUser(t, "bob.tpds")
153
-
alice := p1.MustNewUser(t, "alice.tpds")
154
-
fmt.Println("bob:", bob.DID())
155
-
fmt.Println("alice:", alice.DID())
156
-
157
-
bp1 := bob.Post(t, "cats for cats")
158
-
ap1 := alice.Post(t, "no i like dogs")
159
-
_ = bp1
160
-
_ = ap1
161
-
162
-
xrpcc := xrpc.Client{
163
-
Host: "http://localhost:7711",
164
-
Client: util.TestingHTTPClient(),
165
-
}
166
-
167
-
// no auth required
168
-
queryOut, err := comatproto.LabelQueryLabels(ctx, &xrpcc, "", 20, []string{}, []string{"*"})
169
-
assert.NoError(err)
170
-
assert.Equal(0, len(queryOut.Labels))
171
-
assert.Nil(queryOut.Cursor)
172
-
173
-
// TODO: many more tests
174
-
}