diff --git a/pkg/api/stream_key.go b/pkg/api/stream_key.go index 3377fb10c..27e77f94e 100644 --- a/pkg/api/stream_key.go +++ b/pkg/api/stream_key.go @@ -77,6 +77,14 @@ func (a *StreamplaceAPI) MakeMediaSigner(ctx context.Context, keyStr string) (me } ctx = log.WithLogValues(ctx, "did", did) + labels, err := a.Model.GetActiveLabels(did) + if err != nil { + return nil, fmt.Errorf("failed to get active labels: %w", err) + } + if atproto.IsBanned(labels...) { + log.Error(ctx, "user is banned", "did", did) + return nil, fmt.Errorf("user is banned") + } var mediaSigner media.MediaSigner if a.CLI.ExternalSigning { diff --git a/pkg/atproto/labeler_firehose.go b/pkg/atproto/labeler_firehose.go index dae283c0f..f537c12ca 100644 --- a/pkg/atproto/labeler_firehose.go +++ b/pkg/atproto/labeler_firehose.go @@ -10,8 +10,10 @@ import ( comatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/label" + "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/events" "github.com/bluesky-social/indigo/events/schedulers/parallel" + "github.com/bluesky-social/indigo/util" "github.com/gorilla/websocket" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/aqhttp" @@ -108,7 +110,6 @@ func (atsync *ATProtoSynchronizer) StartLabelerFirehoseRetry(ctx context.Context rsc := &events.RepoStreamCallbacks{ LabelLabels: func(evt *comatproto.LabelSubscribeLabels_Labels) error { - log.Log(ctx, "labeler labels", "labels", evt.Labels, "seq", evt.Seq) err = atsync.Model.UpdateLabelerCursor(did, evt.Seq) if err != nil { log.Error(ctx, "failed to update labeler cursor", "err", err) @@ -131,22 +132,66 @@ func (atsync *ATProtoSynchronizer) StartLabelerFirehoseRetry(ctx context.Context log.Error(ctx, "failed to marshal label", "err", err) continue } + cts, err := time.Parse(util.ISO8601, l.CreatedAt) + if err != nil { + log.Error(ctx, "failed to parse label created time", "err", err) + continue + } + var exp time.Time + if l.ExpiresAt != nil { + e, err := time.Parse(util.ISO8601, *l.ExpiresAt) + if err != nil { + log.Error(ctx, "failed to parse label expiration time", "err", err) + continue + } + exp = e.UTC() + } else { + exp = time.Time{} + } + neg := false + if l.Negated != nil { + neg = *l.Negated + } + + var targetDID string + + // the URI can either be a true URI or a DID, so + aturi, err1 := syntax.ParseATURI(l.URI) + if err1 != nil { + did, err2 := syntax.ParseDID(l.URI) + if err2 != nil { + log.Error(ctx, "failed to parse label URI as either ATURI or DID", "err1", err1, "err2", err2) + continue + } + targetDID = did.String() + } else { + did, err := aturi.Authority().AsDID() + if err != nil { + log.Error(ctx, "failed to parse label URI as ATURI", "err", err) + continue + } + targetDID = did.String() + } + + log.Log(ctx, "labeler label", "targetDID", targetDID, "uri", l.URI, "cid", l.CID, "createdAt", cts, "expiresAt", exp, "negated", neg, "sourceDID", l.SourceDID, "val", l.Val, "version", l.Version) err = atsync.Model.CreateLabel(&model.Label{ - Cid: l.CID, - Cts: l.CreatedAt, - Exp: l.ExpiresAt, - Neg: l.Negated, - Sig: l.Sig, - Src: l.SourceDID, - Uri: l.URI, - Val: l.Val, - Ver: &l.Version, - Record: bs.Bytes(), + Cid: l.CID, + Cts: cts.UTC(), + Exp: exp, + Neg: neg, + Sig: l.Sig, + Src: l.SourceDID, + Uri: l.URI, + Val: l.Val, + Ver: &l.Version, + Record: bs.Bytes(), + RepoDID: targetDID, }) if err != nil { log.Error(ctx, "failed to create label", "err", err) continue } + atsync.Bus.Publish(targetDID, labelLex) } return nil }, diff --git a/pkg/atproto/labels.go b/pkg/atproto/labels.go new file mode 100644 index 000000000..350f704a5 --- /dev/null +++ b/pkg/atproto/labels.go @@ -0,0 +1,52 @@ +package atproto + +import ( + "strings" + + comatproto "github.com/bluesky-social/indigo/api/atproto" +) + +const ( + // from com.atproto.label.defs + LabelHide = "!hide" + LabelNoPromote = "!no-promote" + LabelWarn = "!warn" + LabelNoUnauthenticated = "!no-unauthenticated" + LabelDMCAViolation = "dmca-violation" + LabelDoxxing = "doxxing" + LabelPorn = "porn" + LabelSexual = "sexual" + LabelNudity = "nudity" + LabelNSFL = "nsfl" + LabelGore = "gore" + + // referenced in https://atproto.com/specs/label + LabelTakedown = "!takedown" + LabelSuspend = "!suspend" +) + +var bannedLabels = map[string]bool{ + LabelDMCAViolation: true, + LabelDoxxing: true, + LabelPorn: true, + LabelSexual: true, + LabelNudity: true, + LabelNSFL: true, + LabelGore: true, + LabelTakedown: true, + LabelSuspend: true, +} + +// Given a users' labels, determine if they are banned +func IsBanned(labels ...*comatproto.LabelDefs_Label) bool { + for _, l := range labels { + if !strings.HasPrefix(l.Uri, "did:") { + // this is a label on a record, not a user + continue + } + if bannedLabels[l.Val] { + return true + } + } + return false +} diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go index 68d3fc5e8..aa924933a 100644 --- a/pkg/bus/bus.go +++ b/pkg/bus/bus.go @@ -62,7 +62,11 @@ func (b *Bus) Unsubscribe(user string, ch <-chan Message) { func (b *Bus) Publish(user string, msg Message) { b.mu.Lock() defer b.mu.Unlock() - for _, sub := range b.clients[user] { + subs, ok := b.clients[user] + if !ok { + return + } + for _, sub := range subs { go func(sub Subscription) { sub <- msg }(sub) diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index e768b71f4..1f73882aa 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -365,11 +365,11 @@ func start(build *config.BuildFlags, platformJobs []jobFunc) error { group.Go(func() error { return atsync.StartFirehose(ctx) }) - for _, labeler := range cli.Labelers { - group.Go(func() error { - return atsync.StartLabelerFirehose(ctx, labeler) - }) - } + } + for _, labeler := range cli.Labelers { + group.Go(func() error { + return atsync.StartLabelerFirehose(ctx, labeler) + }) } group.Go(func() error { diff --git a/pkg/media/key_revocation.go b/pkg/media/key_revocation.go index c84865431..9f1949f23 100644 --- a/pkg/media/key_revocation.go +++ b/pkg/media/key_revocation.go @@ -4,11 +4,13 @@ import ( "context" "fmt" + comatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/go-gst/go-gst/gst" + "stream.place/streamplace/pkg/atproto" "stream.place/streamplace/pkg/model" ) -// Handle shutting down a pipeline when a signing key is revoked +// Handle shutting down a pipeline when a signing key is revoked or a user gets banned func (mm *MediaManager) HandleKeyRevocation(ctx context.Context, ms MediaSigner, pipeline *gst.Pipeline) { sub := mm.bus.Subscribe(ms.Streamer()) defer mm.bus.Unsubscribe(ms.Streamer(), sub) @@ -17,18 +19,25 @@ func (mm *MediaManager) HandleKeyRevocation(ctx context.Context, ms MediaSigner, case <-ctx.Done(): return case msg := <-sub: - signingKey, ok := msg.(*model.SigningKey) - if !ok { + switch v := msg.(type) { + case *model.SigningKey: + if v.RevokedAt == nil { + continue + } + if v.DID == ms.DID() { + err := fmt.Errorf("signing key revoked, ending stream: %s", v.RKey) + pipeline.Error(err.Error(), err) + return + } + case *comatproto.LabelDefs_Label: + if atproto.IsBanned(v) { + err := fmt.Errorf("user banned, ending stream: %s", v.Uri) + pipeline.Error(err.Error(), err) + return + } + default: continue } - if signingKey.RevokedAt == nil { - continue - } - if signingKey.DID == ms.DID() { - err := fmt.Errorf("signing key revoked, ending stream: %s", signingKey.RKey) - pipeline.Error(err.Error(), err) - return - } } } } diff --git a/pkg/model/label.go b/pkg/model/label.go index 33f2c3e3d..14555fe9e 100644 --- a/pkg/model/label.go +++ b/pkg/model/label.go @@ -1,16 +1,22 @@ package model -import "gorm.io/gorm/clause" +import ( + "bytes" + "time" + + comatproto "github.com/bluesky-social/indigo/api/atproto" + "gorm.io/gorm/clause" +) type Label struct { // cid: Optionally, CID specifying the specific version of 'uri' resource this label applies to. Cid *string `json:"cid,omitempty" cborgen:"cid,omitempty" gorm:"column:cid"` // cts: Timestamp when this label was created. - Cts string `json:"cts" cborgen:"cts" gorm:"column:cts"` + Cts time.Time `json:"cts" cborgen:"cts" gorm:"column:cts"` // exp: Timestamp at which this label expires (no longer applies). - Exp *string `json:"exp,omitempty" cborgen:"exp,omitempty" gorm:"column:exp"` + Exp time.Time `json:"exp,omitempty" cborgen:"exp,omitempty" gorm:"column:exp"` // neg: If true, this is a negation label, overwriting a previous label. - Neg *bool `json:"neg,omitempty" cborgen:"neg,omitempty" gorm:"column:neg"` + Neg bool `json:"neg,omitempty" cborgen:"neg,omitempty" gorm:"column:neg"` // sig: Signature of dag-cbor encoded label. Sig []byte `json:"sig,omitempty" cborgen:"sig,omitempty" gorm:"column:sig"` // src: DID of the actor who created this label. @@ -22,7 +28,8 @@ type Label struct { // ver: The AT Protocol version of the label object. Ver *int64 `json:"ver,omitempty" cborgen:"ver,omitempty" gorm:"column:ver"` - Record []byte `json:"record,omitempty" cborgen:"record,omitempty" gorm:"column:record"` + Record []byte `json:"record,omitempty" cborgen:"record,omitempty" gorm:"column:record"` + RepoDID string `json:"repoDID,omitempty" cborgen:"repoDID,omitempty" gorm:"column:repo_did"` } func (m *DBModel) CreateLabel(label *Label) error { @@ -35,3 +42,31 @@ func (m *DBModel) CreateLabel(label *Label) error { UpdateAll: true, }).Create(label).Error } + +func (m *DBModel) GetActiveLabels(uri string) ([]*comatproto.LabelDefs_Label, error) { + now := time.Now().UTC() + var labels []Label + err := m.DB.Where("uri = ? AND (exp IS NULL OR exp < ?) AND neg = ?", uri, now, false).Find(&labels).Error + if err != nil { + return nil, err + } + lexs := make([]*comatproto.LabelDefs_Label, len(labels)) + for i, l := range labels { + lex, err := l.ToLexicon() + if err != nil { + return nil, err + } + lexs[i] = lex + } + return lexs, nil +} + +func (l Label) ToLexicon() (*comatproto.LabelDefs_Label, error) { + r := bytes.NewReader(l.Record) + var lex comatproto.LabelDefs_Label + err := lex.UnmarshalCBOR(r) + if err != nil { + return nil, err + } + return &lex, nil +} diff --git a/pkg/model/model.go b/pkg/model/model.go index 38d1ca34c..46bb8d88f 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -113,6 +113,7 @@ type Model interface { UpdateLabelerCursor(did string, cursor int64) error CreateLabel(label *Label) error + GetActiveLabels(uri string) ([]*comatproto.LabelDefs_Label, error) } func MakeDB(dbURL string) (Model, error) {