From f4682b9cd42e027e102afbf593580c480455b4f2 Mon Sep 17 00:00:00 2001 From: Alan Johnson Date: Sun, 2 Aug 2026 11:32:12 -0400 Subject: [PATCH] Wire RFC 9421 inbound runtime verification --- TODO.md | 6 +- api/api.go | 15 +- api/decode.go | 253 +++++- api/operational_metrics.go | 67 ++ api/operational_metrics_test.go | 32 + api/rfc9421_inbound.go | 113 +++ api/rfc9421_inbound_test.go | 271 ++++++ contrib/ops/rfc9421inbound/main.go | 847 ++++++++++++++++++ contrib/ops/test_rfc9421_inbound_probe.sh | 110 +++ docs/HTTP-MESSAGE-SIGNATURES.md | 40 + docs/INTEGRATION-TESTING.md | 31 + internal/httpsignature/rfc9421_verify.go | 14 +- internal/httpsignature/rfc9421_verify_test.go | 8 +- internal/observability/operational.go | 58 ++ internal/observability/operational_test.go | 24 +- 15 files changed, 1826 insertions(+), 63 deletions(-) create mode 100644 api/rfc9421_inbound.go create mode 100644 api/rfc9421_inbound_test.go create mode 100644 contrib/ops/rfc9421inbound/main.go create mode 100755 contrib/ops/test_rfc9421_inbound_probe.sh diff --git a/TODO.md b/TODO.md index c3adbb3..19d54cb 100644 --- a/TODO.md +++ b/TODO.md @@ -71,9 +71,9 @@ the retrospective FEP-ae0c relay profiles. - [x] Add the inbound RFC 9421 verification core with actor/key binding, required component policy, bounded time validation, and Redis-backed nonce replay prevention. -- [ ] Wire the inbound core into the ActivityPub decoder with authenticated - actor-key retrieval, legacy/modern profile selection, metrics, and a - real-process signed inbound probe. +- [x] Wire the inbound core into the ActivityPub decoder with authenticated + actor-key retrieval, strict legacy/modern profile selection, bounded + metrics, and a real-process signed inbound/replay/tamper probe. - [ ] Add explicit outbound `legacy` and `rfc9421` configuration with safe startup validation and no default change. - [ ] Design and test destination-aware `dual` negotiation without ambiguous diff --git a/api/api.go b/api/api.go index dbadb06..cfd7149 100644 --- a/api/api.go +++ b/api/api.go @@ -25,11 +25,12 @@ var ( // WebfingerResources : Relay's Webfinger Resources WebfingerResources []models.WebfingerResource - ActorCache *cache.Cache - MachineryServer *machinery.Server - RemoteRequestSigner *httpsignature.Signer - RelayState models.RelayState - OperationalMetrics *observability.Recorder + ActorCache *cache.Cache + MachineryServer *machinery.Server + RemoteRequestSigner *httpsignature.Signer + InboundRFC9421Verifier *httpsignature.RFC9421Verifier + RelayState models.RelayState + OperationalMetrics *observability.Recorder ) type httpServerBinding struct { @@ -147,6 +148,10 @@ func initialize(globalConfig *models.RelayConfig) error { return err } ActorCache = cache.New(5*time.Minute, 10*time.Minute) + InboundRFC9421Verifier, err = newInboundRFC9421Verifier(globalConfig) + if err != nil { + return err + } Nodeinfo = models.GenerateNodeinfoResources(globalConfig.ServerHostname(), version) WebfingerResources = append(WebfingerResources, RelayActor.GenerateWebfingerResource(globalConfig.ServerHostname())) diff --git a/api/decode.go b/api/decode.go index 18119a4..86aa4a9 100644 --- a/api/decode.go +++ b/api/decode.go @@ -15,67 +15,199 @@ import ( "github.com/thystra/Activity-Relay/models" ) -func decodeActivity(request *http.Request) (*models.Activity, *models.Actor, []byte, error) { - request.Header.Set("Host", request.Host) - body, err := io.ReadAll(io.LimitReader(request.Body, GlobalConfig.MaxActivityBytes()+1)) +const ( + inboundSignatureProfileLegacy = "legacy" + inboundSignatureProfileRFC9421 = "rfc9421" +) + +func selectInboundSignatureProfile(request *http.Request) string { + if request != nil && + len(request.Header.Values("Signature-Input")) != 0 { + return inboundSignatureProfileRFC9421 + } + return inboundSignatureProfileLegacy +} + +func decodeActivity( + request *http.Request, +) (*models.Activity, *models.Actor, []byte, error) { + profile := selectInboundSignatureProfile(request) + body, err := readInboundActivityBody(request) if err != nil { return nil, nil, nil, err } - if int64(len(body)) > GlobalConfig.MaxActivityBytes() { - return nil, nil, nil, errors.New("activity body exceeds configured limit") + + var activity *models.Activity + var actor *models.Actor + + switch profile { + case inboundSignatureProfileRFC9421: + activity, actor, err = decodeRFC9421Activity(request, body) + default: + activity, actor, err = decodeLegacyActivity(request, body) } - // Verify HTTPSignature - verifier, err := httpsig.NewVerifier(request) if err != nil { + recordHTTPSignatureVerification( + profile, + "failure", + classifyHTTPSignatureVerificationError(err), + ) return nil, nil, nil, err } - KeyID := verifier.KeyId() - keyOwnerActor, err := models.NewActivityPubActorFromRemoteActor(KeyID, fmt.Sprintf("%s (golang net/http; Activity-Relay %s; %s)", GlobalConfig.ServerServiceName(), version, GlobalConfig.ServerHostname().Host), ActorCache, RemoteRequestSigner) + + recordHTTPSignatureVerification(profile, "success", "accepted") + return activity, actor, body, nil +} + +func readInboundActivityBody(request *http.Request) ([]byte, error) { + if request == nil || request.Body == nil { + return nil, errors.New("activity request body is unavailable") + } + body, err := io.ReadAll( + io.LimitReader( + request.Body, + GlobalConfig.MaxActivityBytes()+1, + ), + ) if err != nil { - return nil, nil, nil, err + return nil, err + } + if int64(len(body)) > GlobalConfig.MaxActivityBytes() { + return nil, errors.New( + "activity body exceeds configured limit", + ) } - PubKey, err := models.ReadPublicKeyRSAFromString(keyOwnerActor.PublicKey.PublicKeyPem) - if PubKey == nil { - return nil, nil, nil, errors.New("failed parse PublicKey from string") + return body, nil +} + +func parseInboundActivity(body []byte) (*models.Activity, error) { + var activity models.Activity + if err := json.Unmarshal(body, &activity); err != nil { + return nil, err } + if strings.TrimSpace(activity.Actor) == "" { + return nil, errors.New("activity has no actor") + } + return &activity, nil +} + +func decodeRFC9421Activity( + request *http.Request, + body []byte, +) (*models.Activity, *models.Actor, error) { + if InboundRFC9421Verifier == nil { + return nil, nil, errors.New( + "RFC 9421 inbound verifier is unavailable", + ) + } + + verification, err := InboundRFC9421Verifier.VerifyPOST( + request, + body, + ) if err != nil { - return nil, nil, nil, err + return nil, nil, err } - err = verifier.Verify(PubKey, httpsig.RSA_SHA256) + + activity, err := parseInboundActivity(body) if err != nil { - return nil, nil, nil, err + return nil, nil, err + } + if err := verification.BindActivityActor(activity.Actor); err != nil { + return nil, nil, err + } + + remoteActor, err := models.NewActivityPubActorFromRemoteActor( + activity.Actor, + relayRemoteUserAgent(), + ActorCache, + RemoteRequestSigner, + ) + if err != nil { + return nil, nil, err + } + if err := verification.BindActivityActor(remoteActor.ID); err != nil { + return nil, nil, err + } + + return activity, &remoteActor, nil +} + +func decodeLegacyActivity( + request *http.Request, + body []byte, +) (*models.Activity, *models.Actor, error) { + request.Header.Set("Host", request.Host) + + verifier, err := httpsig.NewVerifier(request) + if err != nil { + return nil, nil, err + } + keyID := verifier.KeyId() + + keyOwnerActor, err := models.NewActivityPubActorFromRemoteActor( + keyID, + relayRemoteUserAgent(), + ActorCache, + RemoteRequestSigner, + ) + if err != nil { + return nil, nil, err + } + publicKey, err := models.ReadPublicKeyRSAFromString( + keyOwnerActor.PublicKey.PublicKeyPem, + ) + if err != nil { + return nil, nil, err + } + if publicKey == nil { + return nil, nil, errors.New( + "failed parse PublicKey from string", + ) + } + if err := verifier.Verify(publicKey, httpsig.RSA_SHA256); err != nil { + return nil, nil, err } - // Verify Digest givenDigest := request.Header.Get("Digest") hash := sha256.New() - hash.Write(body) - b := hash.Sum(nil) - calculatedDigest := "SHA-256=" + base64.StdEncoding.EncodeToString(b) - + _, _ = hash.Write(body) + calculatedDigest := "SHA-256=" + + base64.StdEncoding.EncodeToString(hash.Sum(nil)) if givenDigest != calculatedDigest { - return nil, nil, nil, errors.New("digest header is mismatch") + return nil, nil, errors.New("digest header is mismatch") } - // Parse Activity - var activity models.Activity - err = json.Unmarshal(body, &activity) + activity, err := parseInboundActivity(body) if err != nil { - return nil, nil, nil, err + return nil, nil, err } - if err := verifySignatureActorBinding(KeyID, keyOwnerActor, activity.Actor); err != nil { - return nil, nil, nil, err + if err := verifySignatureActorBinding( + keyID, + keyOwnerActor, + activity.Actor, + ); err != nil { + return nil, nil, err } - remoteActor, err := models.NewActivityPubActorFromRemoteActor(activity.Actor, fmt.Sprintf("%s (golang net/http; Activity-Relay %s; %s)", GlobalConfig.ServerServiceName(), version, GlobalConfig.ServerHostname().Host), ActorCache, RemoteRequestSigner) + + remoteActor, err := models.NewActivityPubActorFromRemoteActor( + activity.Actor, + relayRemoteUserAgent(), + ActorCache, + RemoteRequestSigner, + ) if err != nil { - return nil, nil, nil, err + return nil, nil, err } - if err := verifyActivityActorDocument(activity.Actor, remoteActor); err != nil { - return nil, nil, nil, err + if err := verifyActivityActorDocument( + activity.Actor, + remoteActor, + ); err != nil { + return nil, nil, err } - return &activity, &remoteActor, body, nil + return activity, &remoteActor, nil } func normalizedRemoteHost(address string) (string, error) { @@ -83,14 +215,20 @@ func normalizedRemoteHost(address string) (string, error) { if err != nil { return "", err } - host := strings.ToLower(strings.TrimSuffix(parsed.Hostname(), ".")) + host := strings.ToLower( + strings.TrimSuffix(parsed.Hostname(), "."), + ) if host == "" { return "", errors.New("remote URL has no host") } return host, nil } -func verifySignatureActorBinding(keyID string, keyOwner models.Actor, activityActor string) error { +func verifySignatureActorBinding( + keyID string, + keyOwner models.Actor, + activityActor string, +) error { actorHost, err := normalizedRemoteHost(activityActor) if err != nil { return fmt.Errorf("invalid activity actor: %w", err) @@ -100,43 +238,68 @@ func verifySignatureActorBinding(keyID string, keyOwner models.Actor, activityAc return fmt.Errorf("invalid signature key ID: %w", err) } if keyHost != actorHost { - return errors.New("signature key host does not match activity actor host") + return errors.New( + "signature key host does not match activity actor host", + ) } if keyOwner.ID != "" { ownerHost, err := normalizedRemoteHost(keyOwner.ID) if err != nil || ownerHost != actorHost { - return errors.New("signature key owner does not match activity actor host") + return errors.New( + "signature key owner does not match activity actor host", + ) } } if keyOwner.PublicKey.Owner != "" { - ownerHost, err := normalizedRemoteHost(keyOwner.PublicKey.Owner) + ownerHost, err := normalizedRemoteHost( + keyOwner.PublicKey.Owner, + ) if err != nil || ownerHost != actorHost { - return errors.New("public key owner does not match activity actor host") + return errors.New( + "public key owner does not match activity actor host", + ) } } return nil } -func verifyActivityActorDocument(activityActor string, remoteActor models.Actor) error { +func verifyActivityActorDocument( + activityActor string, + remoteActor models.Actor, +) error { activityHost, err := normalizedRemoteHost(activityActor) if err != nil { return fmt.Errorf("invalid activity actor: %w", err) } remoteHost, err := normalizedRemoteHost(remoteActor.ID) if err != nil || remoteHost != activityHost { - return errors.New("resolved actor document does not match activity actor host") + return errors.New( + "resolved actor document does not match activity actor host", + ) } return nil } -func fetchOriginalActivityFromURL(url string) (*models.Activity, *models.Actor, error) { - remoteActivity, err := models.NewActivityPubActivityFromRemoteActivity(url, fmt.Sprintf("%s (golang net/http; Activity-Relay %s; %s)", GlobalConfig.ServerServiceName(), version, GlobalConfig.ServerHostname().Host), RemoteRequestSigner) +func fetchOriginalActivityFromURL( + address string, +) (*models.Activity, *models.Actor, error) { + remoteActivity, err := + models.NewActivityPubActivityFromRemoteActivity( + address, + relayRemoteUserAgent(), + RemoteRequestSigner, + ) if err != nil { return nil, nil, err } - remoteActor, err := models.NewActivityPubActorFromRemoteActor(remoteActivity.Actor, fmt.Sprintf("%s (golang net/http; Activity-Relay %s; %s)", GlobalConfig.ServerServiceName(), version, GlobalConfig.ServerHostname().Host), ActorCache, RemoteRequestSigner) + remoteActor, err := models.NewActivityPubActorFromRemoteActor( + remoteActivity.Actor, + relayRemoteUserAgent(), + ActorCache, + RemoteRequestSigner, + ) if err != nil { return &remoteActivity, nil, err } - return &remoteActivity, &remoteActor, err + return &remoteActivity, &remoteActor, nil } diff --git a/api/operational_metrics.go b/api/operational_metrics.go index 12d3b33..c3af843 100644 --- a/api/operational_metrics.go +++ b/api/operational_metrics.go @@ -2,10 +2,12 @@ package api import ( "context" + "errors" "net/http" "strings" "github.com/sirupsen/logrus" + relayhttpsig "github.com/thystra/Activity-Relay/internal/httpsignature" "github.com/thystra/Activity-Relay/internal/observability" "github.com/thystra/Activity-Relay/models" ) @@ -55,6 +57,71 @@ func recordActivityMetric(result, activityType, reason string) { logrus.WithError(err).Debug("Unable to record activity metric") } } +func recordHTTPSignatureVerification(profile, result, reason string) { + if OperationalMetrics == nil { + return + } + if err := OperationalMetrics.RecordHTTPSignatureVerification( + context.Background(), + profile, + result, + reason, + ); err != nil { + logrus.WithError(err).Debug( + "Unable to record HTTP signature verification metric", + ) + } +} + +func classifyHTTPSignatureVerificationError(err error) string { + if err == nil { + return "accepted" + } + if errors.Is(err, relayhttpsig.ErrRFC9421Replay) { + return "replay" + } + + text := strings.ToLower(err.Error()) + switch { + case strings.Contains(text, "content-digest") || + strings.Contains(text, "digest header"): + return "digest" + case strings.Contains(text, "older than") || + strings.Contains(text, "future") || + strings.Contains(text, "expired") || + strings.Contains(text, "expires") || + strings.Contains(text, "created parameter"): + return "time" + case strings.Contains(text, "authority"): + return "authority" + case strings.Contains(text, "actor") || + strings.Contains(text, "owner"): + return "actor" + case strings.Contains(text, "resolve") || + strings.Contains(text, "public key") || + strings.Contains(text, "keyid"): + return "key" + case strings.Contains(text, "reserve rfc 9421 nonce") || + strings.Contains(text, "redis"): + return "redis" + case strings.Contains(text, "covered component") || + strings.Contains(text, "algorithm") || + strings.Contains(text, "nonce parameter"): + return "policy" + case strings.Contains(text, "parse rfc 9421") || + strings.Contains(text, "find activitypub") || + strings.Contains(text, "structured field"): + return "parse" + case strings.Contains(text, "verify rfc 9421 signature") || + strings.Contains(text, "verification error"): + return "crypto" + case strings.Contains(text, "invalid character") || + strings.Contains(text, "activity has no actor"): + return "activity" + default: + return "other" + } +} func recordQueueAdmission(kind, result, reason string) { if OperationalMetrics == nil { diff --git a/api/operational_metrics_test.go b/api/operational_metrics_test.go index 5b9b731..e4b5711 100644 --- a/api/operational_metrics_test.go +++ b/api/operational_metrics_test.go @@ -8,6 +8,7 @@ import ( "strings" "testing" + relayhttpsig "github.com/thystra/Activity-Relay/internal/httpsignature" "github.com/thystra/Activity-Relay/internal/observability" "github.com/thystra/Activity-Relay/models" ) @@ -70,3 +71,34 @@ func TestEnqueueActivityRecordsNoTargets(t *testing.T) { t.Fatalf("no-target queue metric missing: %#v", ledger) } } + +func TestClassifyHTTPSignatureVerificationError(t *testing.T) { + tests := []struct { + err error + want string + }{ + {nil, "accepted"}, + {relayhttpsig.ErrRFC9421Replay, "replay"}, + {errors.New("Content-Digest mismatch"), "digest"}, + {errors.New("signature is older than allowed"), "time"}, + {errors.New("request authority mismatch"), "authority"}, + {errors.New("public-key owner mismatch"), "actor"}, + {errors.New("resolve RFC 9421 key"), "key"}, + {errors.New("reserve RFC 9421 nonce: Redis unavailable"), "redis"}, + {errors.New("required covered component is missing"), "policy"}, + {errors.New("parse RFC 9421 signature fields"), "parse"}, + {errors.New("verify RFC 9421 signature"), "crypto"}, + {errors.New("invalid character in activity"), "activity"}, + {errors.New("opaque"), "other"}, + } + for _, test := range tests { + if got := classifyHTTPSignatureVerificationError(test.err); got != test.want { + t.Errorf( + "classifyHTTPSignatureVerificationError(%v) = %q; want %q", + test.err, + got, + test.want, + ) + } + } +} diff --git a/api/rfc9421_inbound.go b/api/rfc9421_inbound.go new file mode 100644 index 0000000..c8c7316 --- /dev/null +++ b/api/rfc9421_inbound.go @@ -0,0 +1,113 @@ +package api + +import ( + "context" + "errors" + "fmt" + "strings" + + "github.com/thystra/Activity-Relay/internal/httpsignature" + "github.com/thystra/Activity-Relay/models" +) + +type activityPubRFC9421KeyResolver struct{} + +func relayRemoteUserAgent() string { + return fmt.Sprintf( + "%s (golang net/http; Activity-Relay %s; %s)", + GlobalConfig.ServerServiceName(), + version, + GlobalConfig.ServerHostname().Host, + ) +} + +func (activityPubRFC9421KeyResolver) ResolveRFC9421Key( + _ context.Context, + keyID string, +) (httpsignature.RFC9421ResolvedKey, error) { + keyID = strings.TrimSpace(keyID) + if keyID == "" { + return httpsignature.RFC9421ResolvedKey{}, + errors.New("RFC 9421 key ID is empty") + } + if ActorCache == nil { + return httpsignature.RFC9421ResolvedKey{}, + errors.New("ActivityPub actor cache is unavailable") + } + if RemoteRequestSigner == nil { + return httpsignature.RFC9421ResolvedKey{}, + errors.New("remote request signer is unavailable") + } + + actor, err := models.NewActivityPubActorFromRemoteActor( + keyID, + relayRemoteUserAgent(), + ActorCache, + RemoteRequestSigner, + ) + if err != nil { + return httpsignature.RFC9421ResolvedKey{}, err + } + if strings.TrimSpace(actor.ID) == "" { + return httpsignature.RFC9421ResolvedKey{}, + errors.New("resolved ActivityPub actor has no ID") + } + if strings.TrimSpace(actor.PublicKey.ID) == "" { + return httpsignature.RFC9421ResolvedKey{}, + errors.New("resolved ActivityPub actor has no public-key ID") + } + if actor.PublicKey.ID != keyID { + return httpsignature.RFC9421ResolvedKey{}, + errors.New("resolved ActivityPub public-key ID does not match keyid") + } + if strings.TrimSpace(actor.PublicKey.Owner) == "" { + return httpsignature.RFC9421ResolvedKey{}, + errors.New("resolved ActivityPub public key has no owner") + } + + publicKey, err := models.ReadPublicKeyRSAFromString( + actor.PublicKey.PublicKeyPem, + ) + if err != nil { + return httpsignature.RFC9421ResolvedKey{}, + fmt.Errorf("parse resolved ActivityPub RSA public key: %w", err) + } + if publicKey == nil { + return httpsignature.RFC9421ResolvedKey{}, + errors.New("resolved ActivityPub RSA public key is nil") + } + + return httpsignature.RFC9421ResolvedKey{ + KeyID: actor.PublicKey.ID, + Owner: actor.PublicKey.Owner, + ActorID: actor.ID, + PublicKey: publicKey, + }, nil +} + +func newInboundRFC9421Verifier( + globalConfig *models.RelayConfig, +) (*httpsignature.RFC9421Verifier, error) { + if globalConfig == nil || globalConfig.ServerHostname() == nil { + return nil, errors.New( + "RFC 9421 public server identity is unavailable", + ) + } + + nonceStore, err := httpsignature.NewRedisRFC9421NonceStore( + globalConfig.RedisClient(), + "", + ) + if err != nil { + return nil, err + } + + return httpsignature.NewRFC9421Verifier( + httpsignature.RFC9421VerifierOptions{ + Scheme: globalConfig.ServerHostname().Scheme, + Authority: globalConfig.ServerHostname().Host, + KeyResolver: activityPubRFC9421KeyResolver{}, + NonceStore: nonceStore, + }, + ) +} diff --git a/api/rfc9421_inbound_test.go b/api/rfc9421_inbound_test.go new file mode 100644 index 0000000..a28d81e --- /dev/null +++ b/api/rfc9421_inbound_test.go @@ -0,0 +1,271 @@ +package api + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "encoding/json" + "encoding/pem" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/patrickmn/go-cache" + relayhttpsig "github.com/thystra/Activity-Relay/internal/httpsignature" + "github.com/thystra/Activity-Relay/models" +) + +func runtimeTestPublicKeyPEM( + t *testing.T, + key *rsa.PublicKey, +) string { + t.Helper() + + der, err := x509.MarshalPKIXPublicKey(key) + if err != nil { + t.Fatal(err) + } + return string(pem.EncodeToMemory(&pem.Block{ + Type: "PUBLIC KEY", + Bytes: der, + })) +} + +func runtimeTestOrigin( + t *testing.T, + key *rsa.PrivateKey, + mutate func(*models.Actor), +) (*httptest.Server, string) { + t.Helper() + + var actor models.Actor + server := httptest.NewServer(http.HandlerFunc( + func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/actor" { + http.NotFound(writer, request) + return + } + writer.Header().Set( + "Content-Type", + "application/activity+json", + ) + if err := json.NewEncoder(writer).Encode(&actor); err != nil { + t.Error(err) + } + }, + )) + t.Cleanup(server.Close) + + actorURL := server.URL + "/actor" + actor = models.Actor{ + Context: []string{ + "https://www.w3.org/ns/activitystreams", + "https://w3id.org/security/v1", + }, + ID: actorURL, + Type: "Application", + Inbox: server.URL + "/inbox", + Endpoints: &models.Endpoints{SharedInbox: server.URL + "/inbox"}, + PublicKey: models.PublicKey{ + ID: actorURL + "#main-key", + Owner: actorURL, + PublicKeyPem: runtimeTestPublicKeyPEM(t, &key.PublicKey), + }, + } + if mutate != nil { + mutate(&actor) + } + return server, actorURL +} + +func preserveRFC9421RuntimeGlobals(t *testing.T) { + t.Helper() + + oldActorCache := ActorCache + oldVerifier := InboundRFC9421Verifier + t.Cleanup(func() { + ActorCache = oldActorCache + InboundRFC9421Verifier = oldVerifier + }) +} + +func runtimeTestVerifier(t *testing.T) { + t.Helper() + + nonceStore, err := relayhttpsig.NewRedisRFC9421NonceStore( + RelayState.RedisClient, + "test:api:rfc9421:runtime:", + ) + if err != nil { + t.Fatal(err) + } + InboundRFC9421Verifier, err = relayhttpsig.NewRFC9421Verifier( + relayhttpsig.RFC9421VerifierOptions{ + Scheme: "https", + Authority: "relay.example", + KeyResolver: activityPubRFC9421KeyResolver{}, + NonceStore: nonceStore, + }, + ) + if err != nil { + t.Fatal(err) + } +} + +func runtimeSignedRequest( + t *testing.T, + key *rsa.PrivateKey, + actorURL string, + body []byte, +) *http.Request { + t.Helper() + + request, err := http.NewRequest( + http.MethodPost, + "https://relay.example/inbox", + bytes.NewReader(body), + ) + if err != nil { + t.Fatal(err) + } + request.Host = "relay.example" + request.Header.Set("Content-Type", "application/activity+json") + + signer, err := relayhttpsig.NewSigner( + actorURL+"#main-key", + key, + ) + if err != nil { + t.Fatal(err) + } + if err := signer.SignPOSTWithProfile( + request, + body, + relayhttpsig.ProfileRFC9421, + ); err != nil { + t.Fatal(err) + } + return request +} + +func TestDecodeActivityAcceptsRFC9421AndRejectsReplay( + t *testing.T, +) { + preserveRFC9421RuntimeGlobals(t) + + ctx := context.Background() + if err := RelayState.RedisClient.FlushDB(ctx).Err(); err != nil { + t.Fatal(err) + } + ActorCache = cache.New(5*time.Minute, 10*time.Minute) + + key, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatal(err) + } + _, actorURL := runtimeTestOrigin(t, key, nil) + runtimeTestVerifier(t) + + body, err := json.Marshal(models.Activity{ + Context: "https://www.w3.org/ns/activitystreams", + ID: actorURL + "/activities/one", + Actor: actorURL, + Type: "Like", + Object: "https://example.invalid/object", + To: []string{ + "https://www.w3.org/ns/activitystreams#Public", + }, + }) + if err != nil { + t.Fatal(err) + } + request := runtimeSignedRequest(t, key, actorURL, body) + + activity, actor, gotBody, err := decodeActivity(request) + if err != nil { + t.Fatalf("decode modern activity: %v", err) + } + if activity.Actor != actorURL || actor.ID != actorURL { + t.Fatalf( + "actor binding = activity %q actor %q", + activity.Actor, + actor.ID, + ) + } + if !bytes.Equal(gotBody, body) { + t.Fatal("decoded body changed") + } + + replay := request.Clone(context.Background()) + replay.Body = io.NopCloser(bytes.NewReader(body)) + replay.ContentLength = int64(len(body)) + if _, _, _, err := decodeActivity(replay); !errors.Is( + err, + relayhttpsig.ErrRFC9421Replay, + ) { + t.Fatalf("replay error = %v", err) + } +} + +func TestRFC9421KeyResolverRequiresExactPublicKeyID(t *testing.T) { + preserveRFC9421RuntimeGlobals(t) + + key, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatal(err) + } + _, actorURL := runtimeTestOrigin( + t, + key, + func(actor *models.Actor) { + actor.PublicKey.ID = actor.ID + "#other-key" + }, + ) + + ActorCache = cache.New(5*time.Minute, 10*time.Minute) + _, err = (activityPubRFC9421KeyResolver{}). + ResolveRFC9421Key( + context.Background(), + actorURL+"#main-key", + ) + if err == nil || !strings.Contains(err.Error(), "does not match") { + t.Fatalf("key mismatch error = %v", err) + } +} + +func TestModernProfileDoesNotDowngradeToLegacy(t *testing.T) { + request := httptest.NewRequest( + http.MethodPost, + "https://relay.example/inbox", + strings.NewReader("{}"), + ) + request.Host = "relay.example" + request.Header["Signature-Input"] = []string{""} + if selectInboundSignatureProfile(request) != + inboundSignatureProfileRFC9421 { + t.Fatal("blank Signature-Input field did not select the modern profile") + } + request.Header.Set("Signature-Input", "not-a-structured-field") + request.Header.Set( + "Signature", + `keyId="https://legacy.example/actor#main-key"`, + ) + + if selectInboundSignatureProfile(request) != + inboundSignatureProfileRFC9421 { + t.Fatal("Signature-Input did not select the modern profile") + } + _, _, _, err := decodeActivity(request) + if err == nil || strings.Contains( + err.Error(), + `neither "Signature" nor "Authorization"`, + ) { + t.Fatalf("modern request fell back to legacy: %v", err) + } +} diff --git a/contrib/ops/rfc9421inbound/main.go b/contrib/ops/rfc9421inbound/main.go new file mode 100644 index 0000000..8790c01 --- /dev/null +++ b/contrib/ops/rfc9421inbound/main.go @@ -0,0 +1,847 @@ +// File: contrib/ops/rfc9421inbound/main.go +// +// Real-process RFC 9421 inbound acceptance and replay probe. + +package main + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/rsa" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/json" + "encoding/pem" + "errors" + "flag" + "fmt" + "io" + "math/big" + "net" + "net/http" + "net/http/httputil" + "net/url" + "os" + "os/exec" + "path/filepath" + "strings" + "sync/atomic" + "time" + + legacyhttpsig "github.com/go-fed/httpsig" + "github.com/redis/go-redis/v9" + relayhttpsig "github.com/thystra/Activity-Relay/internal/httpsignature" +) + +type report struct { + SchemaVersion int `json:"schema_version"` + Classification string `json:"classification"` + InfrastructureError string `json:"infrastructure_error,omitempty"` + StartedAt string `json:"started_at"` + FinishedAt string `json:"finished_at"` + RelayActor string `json:"relay_actor"` + OriginActor string `json:"origin_actor"` + ValidStatus int `json:"valid_status"` + ReplayStatus int `json:"replay_status"` + TamperedStatus int `json:"tampered_status"` + ValidAfterTamperStatus int `json:"valid_after_tamper_status"` + SignedActorGETs int64 `json:"signed_actor_gets"` + InvalidActorGETs int64 `json:"invalid_actor_gets"` + NonceMarkerCount int `json:"nonce_marker_count"` + Metrics map[string]string `json:"metrics"` + ProcessPID int `json:"process_pid"` + ConfigPath string `json:"config_path"` + LogPath string `json:"log_path"` +} + +type processState struct { + cancel context.CancelFunc + cmd *exec.Cmd + log *os.File +} + +func (state *processState) close() { + if state == nil { + return + } + if state.cancel != nil { + state.cancel() + } + if state.cmd != nil && state.cmd.Process != nil { + _ = state.cmd.Process.Kill() + _, _ = state.cmd.Process.Wait() + } + if state.log != nil { + _ = state.log.Close() + } +} + +func freePort(ip string) (int, error) { + listener, err := net.Listen("tcp", net.JoinHostPort(ip, "0")) + if err != nil { + return 0, err + } + defer listener.Close() + return listener.Addr().(*net.TCPAddr).Port, nil +} + +func generateKey() (*rsa.PrivateKey, error) { + return rsa.GenerateKey(rand.Reader, 2048) +} + +func writePrivateKey(path string, key *rsa.PrivateKey) error { + data := pem.EncodeToMemory(&pem.Block{ + Type: "RSA PRIVATE KEY", + Bytes: x509.MarshalPKCS1PrivateKey(key), + }) + return os.WriteFile(path, data, 0o600) +} + +func publicKeyPEM(key *rsa.PublicKey) (string, error) { + data, err := x509.MarshalPKIXPublicKey(key) + if err != nil { + return "", err + } + return string(pem.EncodeToMemory(&pem.Block{ + Type: "PUBLIC KEY", + Bytes: data, + })), nil +} + +func generateCA() ( + *x509.Certificate, + *rsa.PrivateKey, + []byte, + error, +) { + key, err := generateKey() + if err != nil { + return nil, nil, nil, err + } + now := time.Now().Add(-time.Minute) + template := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "RFC 9421 inbound probe CA"}, + NotBefore: now, + NotAfter: now.Add(24 * time.Hour), + IsCA: true, + BasicConstraintsValid: true, + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + } + der, err := x509.CreateCertificate( + rand.Reader, + template, + template, + &key.PublicKey, + key, + ) + if err != nil { + return nil, nil, nil, err + } + certificate, err := x509.ParseCertificate(der) + if err != nil { + return nil, nil, nil, err + } + return certificate, key, pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", + Bytes: der, + }), nil +} + +func generateLeaf( + ca *x509.Certificate, + caKey *rsa.PrivateKey, + ip net.IP, + serial int64, +) (tls.Certificate, error) { + key, err := generateKey() + if err != nil { + return tls.Certificate{}, err + } + now := time.Now().Add(-time.Minute) + template := &x509.Certificate{ + SerialNumber: big.NewInt(serial), + Subject: pkix.Name{CommonName: ip.String()}, + NotBefore: now, + NotAfter: now.Add(24 * time.Hour), + IPAddresses: []net.IP{ip}, + KeyUsage: x509.KeyUsageDigitalSignature | + x509.KeyUsageKeyEncipherment, + ExtKeyUsage: []x509.ExtKeyUsage{ + x509.ExtKeyUsageServerAuth, + }, + } + der, err := x509.CreateCertificate( + rand.Reader, + template, + ca, + &key.PublicKey, + caKey, + ) + if err != nil { + return tls.Certificate{}, err + } + return tls.X509KeyPair( + pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", + Bytes: der, + }), + pem.EncodeToMemory(&pem.Block{ + Type: "RSA PRIVATE KEY", + Bytes: x509.MarshalPKCS1PrivateKey(key), + }), + ) +} + +func startTLSServer( + address string, + certificate tls.Certificate, + handler http.Handler, +) (*http.Server, net.Listener, error) { + listener, err := net.Listen("tcp", address) + if err != nil { + return nil, nil, err + } + tlsListener := tls.NewListener(listener, &tls.Config{ + MinVersion: tls.VersionTLS12, + Certificates: []tls.Certificate{certificate}, + }) + server := &http.Server{ + Handler: handler, + ReadHeaderTimeout: 5 * time.Second, + ReadTimeout: 15 * time.Second, + WriteTimeout: 30 * time.Second, + } + go func() { + _ = server.Serve(tlsListener) + }() + return server, listener, nil +} + +func startProxy( + publicAddress string, + certificate tls.Certificate, + backendAddress string, +) (*http.Server, net.Listener, error) { + target, err := url.Parse("http://" + backendAddress) + if err != nil { + return nil, nil, err + } + proxy := httputil.NewSingleHostReverseProxy(target) + proxy.Transport = &http.Transport{Proxy: nil} + originalDirector := proxy.Director + proxy.Director = func(request *http.Request) { + incomingHost := request.Host + originalDirector(request) + request.Host = incomingHost + } + return startTLSServer(publicAddress, certificate, proxy) +} + +func startRedis(name string, port int) (*redis.Client, error) { + command := exec.Command( + "docker", + "run", + "--detach", + "--rm", + "--name", + name, + "--publish", + fmt.Sprintf("127.0.0.1:%d:6379", port), + "redis:7-alpine", + ) + output, err := command.CombinedOutput() + if err != nil { + return nil, fmt.Errorf( + "start Redis: %w: %s", + err, + strings.TrimSpace(string(output)), + ) + } + client := redis.NewClient(&redis.Options{ + Addr: fmt.Sprintf("127.0.0.1:%d", port), + }) + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + ctx, cancel := context.WithTimeout( + context.Background(), + time.Second, + ) + err = client.Ping(ctx).Err() + cancel() + if err == nil { + return client, nil + } + time.Sleep(200 * time.Millisecond) + } + _ = client.Close() + _ = exec.Command("docker", "rm", "-f", name).Run() + return nil, errors.New("Redis did not become ready") +} + +func verifyLegacyGET( + request *http.Request, + keyID string, + publicKey *rsa.PublicKey, +) bool { + request.Header.Set("Host", request.Host) + verifier, err := legacyhttpsig.NewVerifier(request) + if err != nil || verifier.KeyId() != keyID { + return false + } + return verifier.Verify( + publicKey, + legacyhttpsig.RSA_SHA256, + ) == nil +} + +func writeConfig( + path string, + keyPath string, + redisPort int, + apiAddress string, + publicAuthority string, + observabilityAddress string, +) error { + content := fmt.Sprintf( + `ACTOR_PEM: %s +REDIS_URL: redis://127.0.0.1:%d +RELAY_BIND: %s +OBSERVABILITY_BIND: %s +RELAY_DOMAIN: %s +RELAY_SERVICENAME: RFC 9421 inbound probe +JOB_CONCURRENCY: 1 +MAX_ACTIVITY_BYTES: 1048576 +MAX_FANOUT_TARGETS: 100 +MAX_QUEUE_JOBS: 1000 +`, + keyPath, + redisPort, + apiAddress, + observabilityAddress, + publicAuthority, + ) + return os.WriteFile(path, []byte(content), 0o600) +} + +func startRelay( + parent context.Context, + binary string, + configPath string, + caPath string, + logPath string, +) (*processState, error) { + ctx, cancel := context.WithCancel(parent) + logFile, err := os.OpenFile( + logPath, + os.O_CREATE|os.O_WRONLY|os.O_TRUNC, + 0o600, + ) + if err != nil { + cancel() + return nil, err + } + command := exec.CommandContext( + ctx, + binary, + "--config", + configPath, + "server", + ) + command.Env = append( + os.Environ(), + "SSL_CERT_FILE="+caPath, + "NO_PROXY=127.0.0.1,127.0.0.2,127.0.0.3,localhost", + "no_proxy=127.0.0.1,127.0.0.2,127.0.0.3,localhost", + "HTTP_PROXY=", + "HTTPS_PROXY=", + "ALL_PROXY=", + "http_proxy=", + "https_proxy=", + "all_proxy=", + ) + command.Stdout = logFile + command.Stderr = logFile + if err := command.Start(); err != nil { + _ = logFile.Close() + cancel() + return nil, err + } + return &processState{ + cancel: cancel, + cmd: command, + log: logFile, + }, nil +} + +func waitForOK(client *http.Client, address string) error { + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + response, err := client.Get(address) + if err == nil { + _, _ = io.Copy(io.Discard, response.Body) + _ = response.Body.Close() + if response.StatusCode == http.StatusOK { + return nil + } + } + time.Sleep(200 * time.Millisecond) + } + return fmt.Errorf("endpoint did not become ready: %s", address) +} + +func cloneSignedRequest( + request *http.Request, + body []byte, +) (*http.Request, error) { + cloned, err := http.NewRequest( + request.Method, + request.URL.String(), + bytes.NewReader(body), + ) + if err != nil { + return nil, err + } + cloned.Host = request.Host + cloned.Header = request.Header.Clone() + return cloned, nil +} + +func send( + client *http.Client, + request *http.Request, +) (int, string, error) { + response, err := client.Do(request) + if err != nil { + return 0, "", err + } + defer response.Body.Close() + body, _ := io.ReadAll(io.LimitReader(response.Body, 4096)) + return response.StatusCode, strings.TrimSpace(string(body)), nil +} + +func signedActivityRequest( + inboxURL string, + actorURL string, + key *rsa.PrivateKey, + id string, +) (*http.Request, []byte, error) { + activity := map[string]interface{}{ + "@context": "https://www.w3.org/ns/activitystreams", + "id": actorURL + "/activities/" + id, + "type": "Like", + "actor": actorURL, + "object": "https://example.invalid/object", + "to": []string{ + "https://www.w3.org/ns/activitystreams#Public", + }, + } + body, err := json.Marshal(activity) + if err != nil { + return nil, nil, err + } + request, err := http.NewRequest( + http.MethodPost, + inboxURL, + bytes.NewReader(body), + ) + if err != nil { + return nil, nil, err + } + request.Header.Set("Content-Type", "application/activity+json") + signer, err := relayhttpsig.NewSigner( + actorURL+"#main-key", + key, + ) + if err != nil { + return nil, nil, err + } + if err := signer.SignPOSTWithProfile( + request, + body, + relayhttpsig.ProfileRFC9421, + ); err != nil { + return nil, nil, err + } + return request, body, nil +} + +func run() (result report, err error) { + var relayBinary string + var evidenceDir string + flag.StringVar(&relayBinary, "relay-binary", "", "relay binary") + flag.StringVar(&evidenceDir, "evidence-dir", "", "private evidence directory") + flag.Parse() + + result.SchemaVersion = 1 + result.StartedAt = time.Now().UTC().Format(time.RFC3339Nano) + result.Classification = "infrastructure_failure" + + defer func() { + result.FinishedAt = time.Now().UTC().Format(time.RFC3339Nano) + if err != nil { + result.InfrastructureError = err.Error() + } + }() + + if relayBinary == "" || evidenceDir == "" { + return result, errors.New( + "--relay-binary and --evidence-dir are required", + ) + } + if !filepath.IsAbs(evidenceDir) { + return result, errors.New("evidence directory must be absolute") + } + if err := os.MkdirAll(evidenceDir, 0o700); err != nil { + return result, err + } + + workdir, err := os.MkdirTemp("", "activity-relay-rfc9421-inbound.") + if err != nil { + return result, err + } + defer os.RemoveAll(workdir) + + relayIP := net.ParseIP("127.0.0.2") + originIP := net.ParseIP("127.0.0.3") + relayPublicPort, err := freePort(relayIP.String()) + if err != nil { + return result, err + } + originPort, err := freePort(originIP.String()) + if err != nil { + return result, err + } + backendPort, err := freePort("127.0.0.1") + if err != nil { + return result, err + } + metricsPort, err := freePort("127.0.0.1") + if err != nil { + return result, err + } + redisPort, err := freePort("127.0.0.1") + if err != nil { + return result, err + } + + ca, caKey, caPEM, err := generateCA() + if err != nil { + return result, err + } + relayCertificate, err := generateLeaf( + ca, + caKey, + relayIP, + 2, + ) + if err != nil { + return result, err + } + originCertificate, err := generateLeaf( + ca, + caKey, + originIP, + 3, + ) + if err != nil { + return result, err + } + caPath := filepath.Join(workdir, "ca.pem") + if err := os.WriteFile(caPath, caPEM, 0o600); err != nil { + return result, err + } + + relayKey, err := generateKey() + if err != nil { + return result, err + } + originKey, err := generateKey() + if err != nil { + return result, err + } + relayKeyPath := filepath.Join(workdir, "relay.pem") + if err := writePrivateKey(relayKeyPath, relayKey); err != nil { + return result, err + } + + relayAuthority := fmt.Sprintf( + "%s:%d", + relayIP.String(), + relayPublicPort, + ) + originAuthority := fmt.Sprintf( + "%s:%d", + originIP.String(), + originPort, + ) + relayActor := "https://" + relayAuthority + "/actor" + originActor := "https://" + originAuthority + "/actor" + result.RelayActor = relayActor + result.OriginActor = originActor + + originPublicKey, err := publicKeyPEM(&originKey.PublicKey) + if err != nil { + return result, err + } + + var signedGETs atomic.Int64 + var invalidGETs atomic.Int64 + originHandler := http.HandlerFunc( + func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/actor" { + http.NotFound(writer, request) + return + } + if verifyLegacyGET( + request, + relayActor+"#main-key", + &relayKey.PublicKey, + ) { + signedGETs.Add(1) + } else { + invalidGETs.Add(1) + } + writer.Header().Set( + "Content-Type", + "application/activity+json", + ) + _ = json.NewEncoder(writer).Encode(map[string]interface{}{ + "@context": []string{ + "https://www.w3.org/ns/activitystreams", + "https://w3id.org/security/v1", + }, + "id": originActor, + "type": "Application", + "inbox": "https://" + originAuthority + "/inbox", + "endpoints": map[string]string{ + "sharedInbox": "https://" + originAuthority + "/inbox", + }, + "publicKey": map[string]string{ + "id": originActor + "#main-key", + "owner": originActor, + "publicKeyPem": originPublicKey, + }, + }) + }, + ) + originServer, originListener, err := startTLSServer( + net.JoinHostPort(originIP.String(), fmt.Sprint(originPort)), + originCertificate, + originHandler, + ) + if err != nil { + return result, err + } + defer originServer.Close() + defer originListener.Close() + + backendAddress := fmt.Sprintf("127.0.0.1:%d", backendPort) + proxyServer, proxyListener, err := startProxy( + net.JoinHostPort(relayIP.String(), fmt.Sprint(relayPublicPort)), + relayCertificate, + backendAddress, + ) + if err != nil { + return result, err + } + defer proxyServer.Close() + defer proxyListener.Close() + + redisName := fmt.Sprintf( + "activity-relay-rfc9421-inbound-%d", + time.Now().UnixNano(), + ) + redisClient, err := startRedis(redisName, redisPort) + if err != nil { + return result, err + } + defer func() { + _ = redisClient.Close() + _ = exec.Command("docker", "rm", "-f", redisName).Run() + }() + + configPath := filepath.Join(evidenceDir, "relay.yml") + logPath := filepath.Join(evidenceDir, "relay-server.log") + result.ConfigPath = "relay.yml" + result.LogPath = "relay-server.log" + if err := writeConfig( + configPath, + relayKeyPath, + redisPort, + backendAddress, + relayAuthority, + fmt.Sprintf("127.0.0.1:%d", metricsPort), + ); err != nil { + return result, err + } + + parent, cancel := context.WithCancel(context.Background()) + defer cancel() + process, err := startRelay( + parent, + relayBinary, + configPath, + caPath, + logPath, + ) + if err != nil { + return result, err + } + defer process.close() + result.ProcessPID = process.cmd.Process.Pid + + roots := x509.NewCertPool() + if !roots.AppendCertsFromPEM(caPEM) { + return result, errors.New("unable to trust generated CA") + } + client := &http.Client{ + Timeout: 15 * time.Second, + Transport: &http.Transport{ + Proxy: nil, + TLSClientConfig: &tls.Config{ + MinVersion: tls.VersionTLS12, + RootCAs: roots, + }, + }, + } + if err := waitForOK(client, relayActor); err != nil { + return result, err + } + + inboxURL := "https://" + relayAuthority + "/inbox" + + first, firstBody, err := signedActivityRequest( + inboxURL, + originActor, + originKey, + "valid-one", + ) + if err != nil { + return result, err + } + result.ValidStatus, _, err = send(client, first) + if err != nil { + return result, err + } + + replay, err := cloneSignedRequest(first, firstBody) + if err != nil { + return result, err + } + result.ReplayStatus, _, err = send(client, replay) + if err != nil { + return result, err + } + + second, secondBody, err := signedActivityRequest( + inboxURL, + originActor, + originKey, + "valid-two", + ) + if err != nil { + return result, err + } + tamperedBody := bytes.Replace( + secondBody, + []byte(`"type":"Like"`), + []byte(`"type":"Create"`), + 1, + ) + tampered, err := cloneSignedRequest(second, tamperedBody) + if err != nil { + return result, err + } + result.TamperedStatus, _, err = send(client, tampered) + if err != nil { + return result, err + } + + validAfterTamper, err := cloneSignedRequest(second, secondBody) + if err != nil { + return result, err + } + result.ValidAfterTamperStatus, _, err = send( + client, + validAfterTamper, + ) + if err != nil { + return result, err + } + + result.SignedActorGETs = signedGETs.Load() + result.InvalidActorGETs = invalidGETs.Load() + + ctx, cancelRedis := context.WithTimeout( + context.Background(), + 5*time.Second, + ) + defer cancelRedis() + + nonceKeys, err := redisClient.Keys( + ctx, + "relay:http-signature:nonce:*", + ).Result() + if err != nil { + return result, err + } + result.NonceMarkerCount = len(nonceKeys) + result.Metrics, err = redisClient.HGetAll( + ctx, + "relay:metrics:operational", + ).Result() + if err != nil { + return result, err + } + + successMetric := result.Metrics["http_signature_verifications_total|rfc9421|success|accepted"] + replayMetric := result.Metrics["http_signature_verifications_total|rfc9421|failure|replay"] + digestMetric := result.Metrics["http_signature_verifications_total|rfc9421|failure|digest"] + + if result.ValidStatus != http.StatusAccepted || + result.ReplayStatus != http.StatusBadRequest || + result.TamperedStatus != http.StatusBadRequest || + result.ValidAfterTamperStatus != http.StatusAccepted || + result.SignedActorGETs < 2 || + result.InvalidActorGETs != 0 || + result.NonceMarkerCount != 2 || + successMetric != "2" || + replayMetric != "1" || + digestMetric != "1" { + result.Classification = "invariant_failure" + return result, nil + } + + result.Classification = "rfc9421_inbound_runtime_pass" + return result, nil +} + +func main() { + result, runErr := run() + + evidenceDir := "" + for index, argument := range os.Args { + if argument == "--evidence-dir" && index+1 < len(os.Args) { + evidenceDir = os.Args[index+1] + } + } + if evidenceDir != "" { + _ = os.MkdirAll(evidenceDir, 0o700) + data, _ := json.MarshalIndent(result, "", " ") + _ = os.WriteFile( + filepath.Join(evidenceDir, "report.json"), + append(data, '\n'), + 0o600, + ) + } + data, _ := json.MarshalIndent(result, "", " ") + fmt.Println(string(data)) + + if runErr != nil || result.Classification != + "rfc9421_inbound_runtime_pass" { + os.Exit(1) + } +} diff --git a/contrib/ops/test_rfc9421_inbound_probe.sh b/contrib/ops/test_rfc9421_inbound_probe.sh new file mode 100755 index 0000000..9059bbc --- /dev/null +++ b/contrib/ops/test_rfc9421_inbound_probe.sh @@ -0,0 +1,110 @@ +#!/usr/bin/env bash +# File: contrib/ops/test_rfc9421_inbound_probe.sh + +set -Eeuo pipefail + +repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +evidence_dir="${1:-}" + +if [[ -z "$evidence_dir" ]]; then + printf 'Usage: %s /absolute/path/to/private-evidence-directory\n' "$0" >&2 + exit 2 +fi + +case "$evidence_dir" in + /*) ;; + *) + printf 'Evidence directory must be absolute.\n' >&2 + exit 2 + ;; +esac + +python3 - "$repo" "$evidence_dir" <<'PY' +from pathlib import Path +import sys + +repo = Path(sys.argv[1]).resolve() +evidence = Path(sys.argv[2]).resolve() +try: + evidence.relative_to(repo) +except ValueError: + pass +else: + raise SystemExit("Evidence directory must be outside the repository") +PY + +for command_name in docker go python3 sha256sum; do + command -v "$command_name" >/dev/null 2>&1 || { + printf 'Required command not found: %s\n' "$command_name" >&2 + exit 1 + } +done + +docker info >/dev/null +mkdir -p -m 0700 "$evidence_dir" + +workdir="$(mktemp -d /tmp/activity-relay-rfc9421-inbound-wrapper.XXXXXX)" +cleanup() { + rm -rf -- "$workdir" +} +trap cleanup EXIT + +binary="$workdir/activity-relay" +( + cd "$repo" + go build -trimpath -o "$binary" . + go run ./contrib/ops/rfc9421inbound \ + --relay-binary "$binary" \ + --evidence-dir "$evidence_dir" +) + +sha256sum "$binary" > "$evidence_dir/relay-binary.sha256" +python3 -m json.tool "$evidence_dir/report.json" + +python3 - "$evidence_dir/report.json" <<'PY' +import json +import sys + +with open(sys.argv[1], encoding="utf-8") as handle: + report = json.load(handle) + +errors = [] +expected = { + "classification": "rfc9421_inbound_runtime_pass", + "valid_status": 202, + "replay_status": 400, + "tampered_status": 400, + "valid_after_tamper_status": 202, + "invalid_actor_gets": 0, + "nonce_marker_count": 2, +} +for field, value in expected.items(): + if report.get(field) != value: + errors.append( + "%s is %r, expected %r" + % (field, report.get(field), value) + ) + +if report.get("signed_actor_gets", 0) < 2: + errors.append("signed_actor_gets must be at least 2") + +metrics = report.get("metrics", {}) +expected_metrics = { + "http_signature_verifications_total|rfc9421|success|accepted": "2", + "http_signature_verifications_total|rfc9421|failure|replay": "1", + "http_signature_verifications_total|rfc9421|failure|digest": "1", +} +for field, value in expected_metrics.items(): + if metrics.get(field) != value: + errors.append( + "metric %s is %r, expected %r" + % (field, metrics.get(field), value) + ) + +if errors: + for error in errors: + print("ERROR:", error, file=sys.stderr) + raise SystemExit(1) + +print("RFC 9421 inbound real-process invariant: PASS") +PY diff --git a/docs/HTTP-MESSAGE-SIGNATURES.md b/docs/HTTP-MESSAGE-SIGNATURES.md index b3e7dec..edd7aaa 100644 --- a/docs/HTTP-MESSAGE-SIGNATURES.md +++ b/docs/HTTP-MESSAGE-SIGNATURES.md @@ -110,3 +110,43 @@ actor URL after scheme and authority case normalization. Runtime actor retrieval, inbox profile selection, metrics, and HTTP response classification remain for the next tranche. + +## Inbound runtime selection + +The public inbox now selects RFC 9421 verification whenever a +`Signature-Input` field is present, including an empty malformed field. Once selected, verification failure is +terminal for that request; Activity-Relay does not fall back to the legacy +parser. Requests without `Signature-Input` continue through the established +legacy verification path. + +The runtime key resolver fetches the key ID as an authenticated ActivityPub +GET and requires: + +- a non-empty actor ID; +- an embedded RSA public key; +- the fetched public-key ID to exactly equal `keyid`; +- a non-empty public-key owner; and +- a parseable RSA public key. + +The core verifier then requires the public-key owner and resolved actor ID to +match the activity actor exactly after URL scheme and authority case +normalization. The activity actor document fetched for normal relay processing +is checked through the same exact binding. + +The verifier is initialized from the relay's configured public HTTPS authority +and existing Redis client. This is additive inbound support; outbound delivery +and authorized fetch still use the legacy profile. + +## Inbound signature metrics + +The private metrics surface exports: + +```text +activity_relay_http_signature_verifications_total +``` + +with bounded `profile`, `result`, and `reason` labels. Profiles are `legacy` +and `rfc9421`. Results are `success` and `failure`. Reasons are bounded to +accepted, parse, key, crypto, digest, replay, time, actor, authority, policy, +Redis, activity, or other. Raw actor IDs, key IDs, nonce values, URLs, and +error text are never metric labels. diff --git a/docs/INTEGRATION-TESTING.md b/docs/INTEGRATION-TESTING.md index 9df100c..f48ff3a 100644 --- a/docs/INTEGRATION-TESTING.md +++ b/docs/INTEGRATION-TESTING.md @@ -239,3 +239,34 @@ The suite verifies: The inbox is not yet wired to select this verifier. A subsequent integration tranche must add an ActivityPub key resolver, dual legacy/modern decoder selection, metrics, and a real-process signed inbound probe. + +## RFC 9421 inbound runtime probe + +Run the real-process invariant with evidence outside the repository: + +```text +contrib/ops/test_rfc9421_inbound_probe.sh /absolute/private/evidence/path +``` + +The probe starts: + +- one real Activity-Relay API process; +- one independent Redis container; +- a locally trusted TLS frontend for the relay; +- a locally trusted signed ActivityPub actor origin; and +- a private observability listener. + +It requires: + +- one valid RFC 9421 request to return 202; +- the identical request replay to return 400; +- a body-tampered request to return 400; +- the original valid request after the tamper attempt to return 202, proving + the invalid body did not consume the nonce; +- at least two valid relay-signed actor GETs; +- zero invalid actor GET signatures; +- exactly two temporary nonce markers; and +- bounded success, replay, and digest metrics with expected counts. + +The evidence directory contains the exact relay configuration, process log, +binary checksum, and machine-readable report. diff --git a/internal/httpsignature/rfc9421_verify.go b/internal/httpsignature/rfc9421_verify.go index 1f34a39..0935ffa 100644 --- a/internal/httpsignature/rfc9421_verify.go +++ b/internal/httpsignature/rfc9421_verify.go @@ -323,6 +323,13 @@ func (verifier *RFC9421Verifier) VerifyPOST( return nil, err } + if err := VerifyRFC9530ContentDigestSHA256( + request.Header.Values("Content-Digest"), + body, + ); err != nil { + return nil, err + } + resolved, err := verifier.keyResolver.ResolveRFC9421Key( request.Context(), message.Input.KeyID, @@ -364,13 +371,6 @@ func (verifier *RFC9421Verifier) VerifyPOST( return nil, fmt.Errorf("verify RFC 9421 signature: %w", err) } - if err := VerifyRFC9530ContentDigestSHA256( - request.Header.Values("Content-Digest"), - body, - ); err != nil { - return nil, err - } - reserved, err := verifier.nonceStore.ReserveRFC9421Nonce( request.Context(), message.Input.KeyID, diff --git a/internal/httpsignature/rfc9421_verify_test.go b/internal/httpsignature/rfc9421_verify_test.go index f9bd544..56bc4b9 100644 --- a/internal/httpsignature/rfc9421_verify_test.go +++ b/internal/httpsignature/rfc9421_verify_test.go @@ -173,12 +173,16 @@ func TestRFC9421VerifierDoesNotConsumeNonceForTamperedBody(t *testing.T) { body := []byte(`{"actor":"https://relay.example/actor","type":"Follow"}`) request := signedInboundVerificationRequest(t, signer, body) - if _, err := verifier.VerifyPOST( + _, err := verifier.VerifyPOST( request, []byte(`{"actor":"https://relay.example/actor","type":"Create"}`), - ); err == nil { + ) + if err == nil { t.Fatal("tampered body unexpectedly verified") } + if !strings.Contains(err.Error(), "Content-Digest") { + t.Fatalf("tampered body error = %v; want Content-Digest rejection", err) + } if store.callCount() != 0 { t.Fatalf( "tampered body consumed %d nonce reservation(s)", diff --git a/internal/observability/operational.go b/internal/observability/operational.go index 02ad587..0f592b1 100644 --- a/internal/observability/operational.go +++ b/internal/observability/operational.go @@ -71,6 +71,24 @@ func (recorder *Recorder) RecordActivity(ctx context.Context, result, activityTy normalizeActivityResult(result), normalizeActivityType(activityType), normalizeActivityReason(reason)) } +// RecordHTTPSignatureVerification records one bounded inbound signature +// profile verification outcome. +func (recorder *Recorder) RecordHTTPSignatureVerification( + ctx context.Context, + profile string, + result string, + reason string, +) error { + return recorder.increment( + ctx, + "http_signature_verifications_total", + 1, + normalizeHTTPSignatureProfile(profile), + normalizeHTTPSignatureResult(result), + normalizeHTTPSignatureReason(reason), + ) +} + // RecordQueueAdmission records one queue-admission decision. func (recorder *Recorder) RecordQueueAdmission(ctx context.Context, kind, result, reason string) error { return recorder.increment(ctx, "queue_admissions_total", 1, @@ -172,6 +190,33 @@ func normalizeActivityType(value string) string { func normalizeActivityReason(value string) string { return boundedValue(value, "other", "accepted", "invalid", "policy", "method", "internal", "unsupported") } +func normalizeHTTPSignatureProfile(value string) string { + return boundedValue(value, "other", "legacy", "rfc9421") +} + +func normalizeHTTPSignatureResult(value string) string { + return boundedValue(value, "other", "success", "failure") +} + +func normalizeHTTPSignatureReason(value string) string { + return boundedValue( + value, + "other", + "accepted", + "parse", + "key", + "crypto", + "digest", + "replay", + "time", + "actor", + "authority", + "policy", + "redis", + "activity", + ) +} + func normalizeQueueKind(value string) string { return boundedValue(value, "other", "register", "relay") } @@ -211,6 +256,7 @@ type OperationalCollector struct { client *redis.Client snapshot RuntimeSnapshotFunc activities *prometheus.Desc + signatures *prometheus.Desc queue *prometheus.Desc fanout *prometheus.Desc delivery *prometheus.Desc @@ -232,6 +278,7 @@ func NewOperationalCollector(client *redis.Client, snapshot RuntimeSnapshotFunc) client: client, snapshot: snapshot, activities: prometheus.NewDesc("activity_relay_activities_total", "Accepted, rejected, and ignored inbox activities.", []string{"result", "type", "reason"}, nil), + signatures: prometheus.NewDesc("activity_relay_http_signature_verifications_total", "Inbound HTTP signature verification outcomes by bounded profile and reason.", []string{"profile", "result", "reason"}, nil), queue: prometheus.NewDesc("activity_relay_queue_admissions_total", "Queue admission decisions.", []string{"kind", "result", "reason"}, nil), fanout: prometheus.NewDesc("activity_relay_fanout_targets_total", "Fan-out targets by enqueue outcome.", []string{"result"}, nil), delivery: prometheus.NewDesc("activity_relay_delivery_attempts_total", "Relay fan-out delivery results.", []string{"result", "error_class"}, nil), @@ -247,6 +294,7 @@ func NewOperationalCollector(client *redis.Client, snapshot RuntimeSnapshotFunc) func (collector *OperationalCollector) Describe(channel chan<- *prometheus.Desc) { channel <- collector.activities + channel <- collector.signatures channel <- collector.queue channel <- collector.fanout channel <- collector.delivery @@ -326,6 +374,16 @@ func (collector *OperationalCollector) collectLedger(channel chan<- prometheus.M case len(parts) == 4 && parts[0] == "activities_total": add(collector.activities, parts[0], value, normalizeActivityResult(parts[1]), normalizeActivityType(parts[2]), normalizeActivityReason(parts[3])) + case len(parts) == 4 && + parts[0] == "http_signature_verifications_total": + add( + collector.signatures, + parts[0], + value, + normalizeHTTPSignatureProfile(parts[1]), + normalizeHTTPSignatureResult(parts[2]), + normalizeHTTPSignatureReason(parts[3]), + ) case len(parts) == 4 && parts[0] == "queue_admissions_total": add(collector.queue, parts[0], value, normalizeQueueKind(parts[1]), normalizeQueueResult(parts[2]), normalizeQueueReason(parts[3])) diff --git a/internal/observability/operational_test.go b/internal/observability/operational_test.go index 08f77df..e14c619 100644 --- a/internal/observability/operational_test.go +++ b/internal/observability/operational_test.go @@ -45,6 +45,14 @@ func TestOperationalRecorderAndCollector(t *testing.T) { if err := recorder.RecordActivity(ctx, "invented", "PrivateType", "raw error text"); err != nil { t.Fatal(err) } + if err := recorder.RecordHTTPSignatureVerification( + ctx, + "rfc9421", + "failure", + "replay", + ); err != nil { + t.Fatal(err) + } if err := recorder.RecordQueueAdmission(ctx, "relay", "accepted", "accepted"); err != nil { t.Fatal(err) } @@ -80,6 +88,7 @@ func TestOperationalRecorderAndCollector(t *testing.T) { for _, expected := range []string{ `activity_relay_activities_total{reason="accepted",result="accepted",type="Create"} 1`, `activity_relay_activities_total{reason="other",result="other",type="other"} 1`, + `activity_relay_http_signature_verifications_total{profile="rfc9421",reason="replay",result="failure"} 1`, `activity_relay_queue_admissions_total{kind="relay",reason="accepted",result="accepted"} 1`, `activity_relay_fanout_targets_total{result="queued"} 3`, `activity_relay_delivery_attempts_total{error_class="http_5xx",result="failure"} 1`, @@ -129,6 +138,7 @@ func TestOperationalCollectorBoundsExistingLedgerLabels(t *testing.T) { if err := client.HSet(ctx, OperationalMetricsKey, map[string]interface{}{ "activities_total|invented|PrivateType|raw.example/path": 2, "activities_total|other|other|other": 3, + "http_signature_verifications_total|private|raw|secret": 4, }).Err(); err != nil { t.Fatal(err) } @@ -146,7 +156,19 @@ func TestOperationalCollectorBoundsExistingLedgerLabels(t *testing.T) { if !strings.Contains(body, `activity_relay_activities_total{reason="other",result="other",type="other"} 5`) { t.Fatalf("bounded ledger values were not combined\n%s", body) } - for _, forbidden := range []string{"invented", "PrivateType", "raw.example"} { + if !strings.Contains( + body, + `activity_relay_http_signature_verifications_total{profile="other",reason="other",result="other"} 4`, + ) { + t.Fatalf("bounded signature ledger values were not exported\n%s", body) + } + for _, forbidden := range []string{ + "invented", + "PrivateType", + "raw.example", + "private", + "secret", + } { if strings.Contains(body, forbidden) { t.Fatalf("collector exposed unbounded ledger label %q", forbidden) }