diff --git a/AGENTS.md b/AGENTS.md index 932a17f..3c0a909 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -40,11 +40,16 @@ internal/ controller.go — NATS subscriptions, startup, periodic ticker drift.go — CheckDrift, handlePush, periodicCheck reconciler.go — TriggerSync, handleDeploymentSucceeded/Failed - federation/ — ActivityPub / ForgeFed (DATA LAYER ONLY — Phase 3F stub) + federation/ — ActivityPub / ForgeFed (fully built — Phase 3F) + actor.go — GetOrCreate, ActorJSON, APID, RSA-2048 key gen + signatures.go — HTTP signature sign/verify (draft-cavage-http-signatures) + inbox.go — Receive, handleFollow (auto-accept), handleAccept, handleUndo + outbox.go — Collection builder, StubCollection + remote.go — FetchActor (cached), DeliverActivity (signed POST) observability/ — Prometheus metrics + health (fully built — Phase 3E) metrics.go — metric definitions, HTTP middleware, NATS watcher health.go — Check() returning HealthStatus (DB ping + NATS) - models/ — XORM structs + 13 migration files + models/ — XORM structs + 14 migration files config/ — ENV-driven config, fails fast on missing secrets events/ — NATS EventBus interface + NATSBus + NoOpBus (Healthy() bool) web/ — //go:embed target for the built React SPA @@ -78,11 +83,9 @@ Logger → RealIP → Recoverer → Metrics → CORS → CSRF → SessionAuth | 3C | Workspaces + secret management (Global → Workspace → Repo → Env) | **Complete** | | 3D | GitOps controller + drift detection + auto-sync | **Complete** | | 3E | Observability (Prometheus `/metrics`, structured `/health`, repo health API) | **Complete** | -| 3F | Federation handlers (ActivityPub inbox/outbox) | Planned | +| 3F | Federation handlers (ActivityPub WebFinger, actor, inbox/outbox, HTTP signatures, Follow/Accept) | **Complete** | | 4 | AI diagnostics, signed artifacts, OCI registry, dep/secret scanning | Planned | -The `domain/federation/` directory is an intentional stub — the data model exists but no HTTP handlers should be wired until Phase 3F. - --- ## Go Conventions @@ -108,7 +111,7 @@ This rule is non-negotiable. It prevents command injection. ### Database - XORM for all DB access. Structs in `internal/models/`. -- Migrations are numbered files in `internal/models/migrations/`. Always add a new file; never edit existing ones. Current highest: **013**. +- Migrations are numbered files in `internal/models/migrations/`. Always add a new file; never edit existing ones. Current highest: **014**. - No raw SQL strings built from user input. ### Events @@ -208,6 +211,11 @@ make lint # go vet + ESLint | `internal/api/handlers/observability.go` | `/health` + `/repos/.../health` handlers | | `internal/api/handlers/environment.go` | Environment + deployment CRUD | | `internal/api/handlers/gitops.go` | GitOps config + drift HTTP endpoints | +| `internal/api/handlers/federation.go` | ActivityPub WebFinger, actor, inbox, outbox, followers/following | +| `internal/domain/federation/actor.go` | Actor lifecycle — GetOrCreate, ActorJSON, key generation | +| `internal/domain/federation/signatures.go` | HTTP signature sign/verify | +| `internal/domain/federation/inbox.go` | Receive + Follow/Accept auto-accept flow | +| `internal/domain/federation/remote.go` | FetchActor (cached remote actors), DeliverActivity | | `internal/api/handlers/secret.go` | Scoped secret management | | `internal/api/handlers/workspace.go` | Workspace + member management | | `internal/api/middleware/audit.go` | Audit log middleware | @@ -224,7 +232,7 @@ make lint # go vet + ESLint ```bash cp .env.example .env # fill SESSION_SECRET and CSRF_SECRET make docker-up # PostgreSQL + NATS via Docker Compose -make migrate # run XORM migrations (currently 013) +make migrate # run XORM migrations (currently 014) make dev # Go :8080 + Vite :5173 ``` diff --git a/CHANGELOG.md b/CHANGELOG.md index 8abfcca..e36f589 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,19 +9,6 @@ Versions follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] -### Planned — Phase 3E (Observability) -- Prometheus metrics endpoint `GET /metrics` -- Structured internal metrics: pipeline duration, queue depth, deployment frequency, error rates -- Health check endpoint `GET /health` returning DB + NATS status -- Environment cards: live health status via HTTP health check polling -- Repo page: error rate and deployment frequency sparklines - -### Planned — Phase 3F (Federation, next) -- ActivityPub inbox/outbox HTTP handlers -- HTTP signature verification middleware -- WebFinger `/.well-known/webfinger` endpoint -- Cross-instance pull requests via ActivityPub activities - ### Planned — Phase 4 (Intelligence + Artifacts) - AI failure diagnosis (pipeline failure root-cause analysis via Claude API) - AI deployment risk scoring @@ -30,6 +17,43 @@ Versions follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). - OCI container registry - Secret scanning (commit-level pattern detection) - Dependency vulnerability scanning +- Cross-instance pull requests (ForgeFed ActivityPub extension) + +--- + +## [0.9.0] — 2026-05-12 + +Phase 3F complete. ForgeBucket is now a first-class ActivityPub node — interoperable with Mastodon, Forgejo, and any fediverse server. + +### Added — ActivityPub Federation (`internal/domain/federation/`) + +- **`GET /.well-known/webfinger`** — resolves `acct:user@domain` to the actor URL; returns `application/jrd+json` +- **`GET /users/{username}`** — returns a JSON-LD actor document (`Person` type) including public key object for HTTP signature verification +- **`POST /users/{username}/inbox`** — receives and dispatches inbound ActivityPub activities; HTTP signature verification enforced in production (skipped in `DEBUG=true` mode for local testing) +- **`GET /users/{username}/outbox`** — serves an `OrderedCollection` (summary on page 0, paginated `OrderedCollectionPage` on page ≥ 1, 20 activities per page) +- **`GET /users/{username}/followers`** — stub `OrderedCollection` (zero items; social graph in Phase 4) +- **`GET /users/{username}/following`** — stub `OrderedCollection` + +### Added — HTTP Signatures (`internal/domain/federation/signatures.go`) +- `Sign(req, keyID, privateKeyPEM)` — signs outgoing HTTP requests with RSA-SHA256; covers `(request-target)`, `host`, and `date` headers +- `Verify(r, db, instanceURL)` — parses `Signature` header, resolves sender's public key (local `FederationActor` first, then network fetch via `FetchActor`), verifies RSA-SHA256 digest + +### Added — Actor Lifecycle (`internal/domain/federation/actor.go`) +- `GetOrCreate` — lazily creates a `FederationActor` for a local user; generates a fresh RSA-2048 key pair and derives `InboxURL`, `OutboxURL`, `APID` from `INSTANCE_URL`; stable across requests +- `ActorJSON` — returns the JSON-LD document shape expected by all ActivityPub clients +- `APID(instanceURL, username)` — canonical `{instanceURL}/users/{username}` helper + +### Added — Follow / Accept Flow (`internal/domain/federation/inbox.go`) +- Incoming `Follow` activities are auto-accepted: remote actor is fetched (or retrieved from cache), an `Accept` activity is signed and delivered to their inbox asynchronously +- Both the inbound `Follow` and outbound `Accept` are persisted to `FederationActivity` for audit + +### Added — Remote Actor Cache (`internal/domain/federation/remote.go`) +- `FetchActor` — HTTP GET with `Accept: application/activity+json`, extracts inbox URL and public key PEM, stores in `RemoteActor` table to avoid repeated fetches +- `DeliverActivity` — marshals activity JSON, signs the request, POSTs to recipient inbox with 15-second timeout + +### Added — Database Models (migration `014_federation`) +- `FederationActivity` — append-only log of all inbound and outbound activities: `ActorAPID`, `Type`, `ObjectJSON`, `Direction` (inbound/outbound), `RemoteActor`, `Published` +- `RemoteActor` — cache for remote actor documents: `APID` (unique), `InboxURL`, `PublicKey`, `FetchedAt` --- @@ -312,7 +336,8 @@ Initial development milestone. Core Git hosting, collaboration, and frontend SPA --- -[Unreleased]: https://github.com/forgeo/forgebucket/compare/v0.8.0...HEAD +[Unreleased]: https://github.com/forgeo/forgebucket/compare/v0.9.0...HEAD +[0.9.0]: https://github.com/forgeo/forgebucket/compare/v0.8.0...v0.9.0 [0.8.0]: https://github.com/forgeo/forgebucket/compare/v0.7.0...v0.8.0 [0.7.0]: https://github.com/forgeo/forgebucket/compare/v0.6.0...v0.7.0 [0.6.0]: https://github.com/forgeo/forgebucket/compare/v0.5.0...v0.6.0 diff --git a/README.md b/README.md index a91e753..acbf7b0 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ ForgeBucket is a self-hosted, federated developer operations platform. Where other Git platforms show you a list of files, ForgeBucket surfaces deployments, pipeline health, environment drift, and operational context directly alongside your code. Repositories are runtime systems. The dashboard is a command center. -**Status:** Active development. Phase 3E (observability) complete. Phase 3F (federation handlers) is next. +**Status:** Active development. Phase 3F (ActivityPub federation) complete. Phase 4 (AI diagnostics + signed artifacts) is next. --- @@ -101,9 +101,16 @@ ForgeBucket is a self-hosted, federated developer operations platform. Where oth ### Federation | Feature | Status | |---------|--------| -| ActivityPub actor model | Done (data layer) | -| Federation handlers / inbox / outbox | Planned (Phase 3F) | -| Cross-instance pull requests | Planned (Phase 3F) | +| ActivityPub actor model | Done | +| WebFinger (`/.well-known/webfinger`) | Done | +| Actor documents (`/users/{username}`) | Done | +| Inbox (receive + HTTP signature verify) | Done | +| Outbox (OrderedCollection, paginated) | Done | +| Followers / Following collections | Done | +| HTTP signatures (draft-cavage-http-signatures) | Done | +| Follow / Accept auto-accept flow | Done | +| RSA-2048 key pair lazy generation | Done | +| Cross-instance pull requests (ForgeFed) | Planned (Phase 4) | --- @@ -148,7 +155,7 @@ ForgeBucket ├── Workspace Service (multi-tenant namespaces — internal/api/handlers/workspace.go) ├── Event Bus (NATS core, NoOp fallback — internal/events/) ├── Audit Log (every mutating request — internal/api/middleware/audit.go) -├── Federation Layer (ActivityPub actors — internal/domain/federation/) ← Phase 3F stub +├── Federation Layer (ActivityPub inbox/outbox, HTTP signatures — internal/domain/federation/) ├── Database (PostgreSQL + XORM — internal/models/) └── Web Frontend (React 18 + TypeScript, //go:embed — web/) ``` @@ -176,7 +183,7 @@ Logger → RealIP → Recoverer → Metrics → CORS → CSRF → SessionAuth | YAML parsing | `gopkg.in/yaml.v3` (workflow definitions) | | Code editing | CodeMirror | | Container | Docker Compose (dev) | -| Federation | ActivityPub / ForgeFed (data layer only) | +| Federation | ActivityPub / ForgeFed (WebFinger, actor, inbox/outbox, HTTP signatures) | --- @@ -242,8 +249,8 @@ ForgeBucket has its own design language — intentionally distinct from GitHub a | Phase 3C | Workspaces + secret management hierarchy (Global → Workspace → Repo → Env) | Done | | Phase 3D | GitOps controller + drift detection + auto-sync | Done | | Phase 3E | Observability (Prometheus `/metrics`, structured `/health`, repo health API) | Done | -| Phase 3F | Federation handlers (ActivityPub inbox/outbox, cross-instance PRs) | Next | -| Phase 4 | AI diagnostics, signed artifacts, OCI registry, secret/dep scanning | Planned | +| Phase 3F | Federation handlers (ActivityPub WebFinger, actor, inbox/outbox, HTTP signatures) | Done | +| Phase 4 | AI diagnostics, signed artifacts, OCI registry, secret/dep scanning | Next | --- diff --git a/internal/api/handlers/federation.go b/internal/api/handlers/federation.go new file mode 100644 index 0000000..2f5c5c9 --- /dev/null +++ b/internal/api/handlers/federation.go @@ -0,0 +1,179 @@ +package handlers + +import ( + "encoding/json" + "io" + "net/http" + "strconv" + "strings" + + "github.com/go-chi/chi/v5" + "xorm.io/xorm" + + "github.com/forgeo/forgebucket/internal/config" + "github.com/forgeo/forgebucket/internal/domain/federation" + "github.com/forgeo/forgebucket/internal/models" +) + +const activityJSONType = "application/activity+json" + +type FederationHandler struct { + db *xorm.Engine + cfg *config.Config +} + +func NewFederationHandler(db *xorm.Engine, cfg *config.Config) *FederationHandler { + return &FederationHandler{db: db, cfg: cfg} +} + +// WebFinger handles GET /.well-known/webfinger?resource=acct:user@domain +func (h *FederationHandler) WebFinger(w http.ResponseWriter, r *http.Request) { + resource := r.URL.Query().Get("resource") + if !strings.HasPrefix(resource, "acct:") { + http.Error(w, "resource must use acct: scheme", http.StatusBadRequest) + return + } + + // acct:username@domain — extract username + acct := strings.TrimPrefix(resource, "acct:") + username := strings.SplitN(acct, "@", 2)[0] + + var user models.User + if found, _ := h.db.Where("username = ?", username).Get(&user); !found { + http.NotFound(w, r) + return + } + + actorURL := federation.APID(h.cfg.InstanceURL, username) + resp := map[string]any{ + "subject": resource, + "links": []map[string]any{ + { + "rel": "self", + "type": activityJSONType, + "href": actorURL, + }, + }, + } + + w.Header().Set("Content-Type", "application/jrd+json") + json.NewEncoder(w).Encode(resp) //nolint:errcheck +} + +// Actor handles GET /users/{username} — returns the JSON-LD actor document. +func (h *FederationHandler) Actor(w http.ResponseWriter, r *http.Request) { + username := chi.URLParam(r, "username") + + var user models.User + if found, _ := h.db.Where("username = ?", username).Get(&user); !found { + http.NotFound(w, r) + return + } + + actor, err := federation.GetOrCreate(h.db, user.ID, username, h.cfg.InstanceURL) + if err != nil { + http.Error(w, "could not get actor", http.StatusInternalServerError) + return + } + + doc := federation.ActorJSON(actor, username, username) + w.Header().Set("Content-Type", activityJSONType) + json.NewEncoder(w).Encode(doc) //nolint:errcheck +} + +// Inbox handles POST /users/{username}/inbox — receive an ActivityPub activity. +func (h *FederationHandler) Inbox(w http.ResponseWriter, r *http.Request) { + username := chi.URLParam(r, "username") + + var user models.User + if found, _ := h.db.Where("username = ?", username).Get(&user); !found { + http.NotFound(w, r) + return + } + + actor, err := federation.GetOrCreate(h.db, user.ID, username, h.cfg.InstanceURL) + if err != nil { + http.Error(w, "internal error", http.StatusInternalServerError) + return + } + + body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) // 1 MB max + if err != nil { + http.Error(w, "could not read body", http.StatusBadRequest) + return + } + + // Verify HTTP signature. In debug mode, skip verification so local testing works. + if !h.cfg.Debug { + if err := federation.Verify(r, h.db, h.cfg.InstanceURL); err != nil { + http.Error(w, "signature verification failed: "+err.Error(), http.StatusUnauthorized) + return + } + } + + if err := federation.Receive(h.db, actor, body); err != nil { + http.Error(w, "could not process activity: "+err.Error(), http.StatusBadRequest) + return + } + + w.WriteHeader(http.StatusAccepted) +} + +// OutboxGet handles GET /users/{username}/outbox — serve the activity collection. +func (h *FederationHandler) OutboxGet(w http.ResponseWriter, r *http.Request) { + username := chi.URLParam(r, "username") + + var user models.User + if found, _ := h.db.Where("username = ?", username).Get(&user); !found { + http.NotFound(w, r) + return + } + + actor, err := federation.GetOrCreate(h.db, user.ID, username, h.cfg.InstanceURL) + if err != nil { + http.Error(w, "internal error", http.StatusInternalServerError) + return + } + + page, _ := strconv.Atoi(r.URL.Query().Get("page")) + coll := federation.Collection(h.db, actor.APID, actor.OutboxURL, page) + + w.Header().Set("Content-Type", activityJSONType) + json.NewEncoder(w).Encode(coll) //nolint:errcheck +} + +// Followers handles GET /users/{username}/followers +func (h *FederationHandler) Followers(w http.ResponseWriter, r *http.Request) { + username := chi.URLParam(r, "username") + var user models.User + if found, _ := h.db.Where("username = ?", username).Get(&user); !found { + http.NotFound(w, r) + return + } + actor, err := federation.GetOrCreate(h.db, user.ID, username, h.cfg.InstanceURL) + if err != nil { + http.Error(w, "internal error", http.StatusInternalServerError) + return + } + coll := federation.StubCollection(actor.APID + "/followers") + w.Header().Set("Content-Type", activityJSONType) + json.NewEncoder(w).Encode(coll) //nolint:errcheck +} + +// Following handles GET /users/{username}/following +func (h *FederationHandler) Following(w http.ResponseWriter, r *http.Request) { + username := chi.URLParam(r, "username") + var user models.User + if found, _ := h.db.Where("username = ?", username).Get(&user); !found { + http.NotFound(w, r) + return + } + actor, err := federation.GetOrCreate(h.db, user.ID, username, h.cfg.InstanceURL) + if err != nil { + http.Error(w, "internal error", http.StatusInternalServerError) + return + } + coll := federation.StubCollection(actor.APID + "/following") + w.Header().Set("Content-Type", activityJSONType) + json.NewEncoder(w).Encode(coll) //nolint:errcheck +} diff --git a/internal/api/router.go b/internal/api/router.go index 5ecc8b4..aa74c52 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -65,6 +65,7 @@ func New(cfg *config.Config, engine *xorm.Engine, store sessions.Store, bus even artifactH := handlers.NewArtifactHandler(engine, artifactRoot) runnerH := handlers.NewRunnerHandler(engine) gitopsH := handlers.NewGitOpsHandler(engine, bus) + fedH := handlers.NewFederationHandler(engine, cfg) envH := handlers.NewEnvironmentHandler(engine, bus) timelineH := handlers.NewTimelineHandler(engine, cfg.RepoRoot) workspaceH := handlers.NewWorkspaceHandler(engine, cfg) @@ -279,6 +280,15 @@ func New(cfg *config.Config, engine *xorm.Engine, store sessions.Store, bus even r.With(auth.Optional).Get("/ws", wsH.Hub) + // ── ActivityPub / federation (root-level, no auth) ──────────────────────── + // Must be registered before the /* catch-all so they are not proxied to Vite. + r.Get("/.well-known/webfinger", fedH.WebFinger) + r.Get("/users/{username}", fedH.Actor) + r.Post("/users/{username}/inbox", fedH.Inbox) + r.Get("/users/{username}/outbox", fedH.OutboxGet) + r.Get("/users/{username}/followers", fedH.Followers) + r.Get("/users/{username}/following", fedH.Following) + // In debug mode proxy non-API routes to the Vite dev server so :8080 works too. // In production the built React app is embedded and served from staticFiles. if cfg.Debug { diff --git a/internal/domain/federation/actor.go b/internal/domain/federation/actor.go new file mode 100644 index 0000000..d04cd65 --- /dev/null +++ b/internal/domain/federation/actor.go @@ -0,0 +1,84 @@ +package federation + +import ( + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "encoding/pem" + "fmt" + "strings" + + "xorm.io/xorm" + + "github.com/forgeo/forgebucket/internal/models" +) + +// APID returns the canonical ActivityPub actor ID for a local username. +func APID(instanceURL, username string) string { + return strings.TrimRight(instanceURL, "/") + "/users/" + username +} + +// GetOrCreate fetches the FederationActor for a user, creating it with a fresh +// RSA-2048 key pair if none exists. Actor URLs are derived from instanceURL. +func GetOrCreate(db *xorm.Engine, userID int64, username, instanceURL string) (*models.FederationActor, error) { + var actor models.FederationActor + if found, _ := db.Where("user_id = ?", userID).Get(&actor); found { + return &actor, nil + } + + priv, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + return nil, fmt.Errorf("generate rsa key: %w", err) + } + + privPEM := pem.EncodeToMemory(&pem.Block{ + Type: "RSA PRIVATE KEY", + Bytes: x509.MarshalPKCS1PrivateKey(priv), + }) + pubDER, err := x509.MarshalPKIXPublicKey(&priv.PublicKey) + if err != nil { + return nil, fmt.Errorf("marshal public key: %w", err) + } + pubPEM := pem.EncodeToMemory(&pem.Block{Type: "PUBLIC KEY", Bytes: pubDER}) + + base := APID(instanceURL, username) + actor = models.FederationActor{ + UserID: userID, + APID: base, + InboxURL: base + "/inbox", + OutboxURL: base + "/outbox", + PublicKey: string(pubPEM), + PrivateKey: string(privPEM), + } + if _, err := db.Insert(&actor); err != nil { + // Race condition: another goroutine may have just created it. + if found, _ := db.Where("user_id = ?", userID).Get(&actor); found { + return &actor, nil + } + return nil, fmt.Errorf("insert actor: %w", err) + } + return &actor, nil +} + +// ActorJSON builds the JSON-LD actor document returned by GET /users/{username}. +func ActorJSON(actor *models.FederationActor, username, displayName string) map[string]any { + return map[string]any{ + "@context": []any{ + "https://www.w3.org/ns/activitystreams", + "https://w3id.org/security/v1", + }, + "id": actor.APID, + "type": "Person", + "preferredUsername": username, + "name": displayName, + "inbox": actor.InboxURL, + "outbox": actor.OutboxURL, + "followers": actor.APID + "/followers", + "following": actor.APID + "/following", + "publicKey": map[string]any{ + "id": actor.APID + "#main-key", + "owner": actor.APID, + "publicKeyPem": actor.PublicKey, + }, + } +} diff --git a/internal/domain/federation/inbox.go b/internal/domain/federation/inbox.go new file mode 100644 index 0000000..1534a8f --- /dev/null +++ b/internal/domain/federation/inbox.go @@ -0,0 +1,113 @@ +package federation + +import ( + "encoding/json" + "fmt" + "log" + "time" + + "xorm.io/xorm" + + "github.com/forgeo/forgebucket/internal/models" +) + +// Receive persists an inbound activity and dispatches it by type. +// The caller is responsible for verifying the HTTP signature before calling this. +func Receive(db *xorm.Engine, localActor *models.FederationActor, body []byte) error { + var activity map[string]any + if err := json.Unmarshal(body, &activity); err != nil { + return fmt.Errorf("parse activity: %w", err) + } + + actType, _ := activity["type"].(string) + actorAPID, _ := activity["actor"].(string) + + entry := &models.FederationActivity{ + ActorAPID: localActor.APID, + Type: actType, + ObjectJSON: string(body), + Direction: "inbound", + RemoteActor: actorAPID, + Published: time.Now().UTC(), + } + db.Insert(entry) //nolint:errcheck + + switch actType { + case "Follow": + return handleFollow(db, localActor, activity, actorAPID) + case "Accept": + handleAccept(db, localActor, activity) + case "Undo": + handleUndo(db, localActor, activity) + default: + log.Printf("federation: received unhandled activity type %q from %s", actType, actorAPID) + } + return nil +} + +// handleFollow auto-accepts all incoming Follow activities and sends an Accept +// back to the sender's inbox. +func handleFollow(db *xorm.Engine, localActor *models.FederationActor, follow map[string]any, followerAPID string) error { + if followerAPID == "" { + return fmt.Errorf("Follow activity missing actor field") + } + + // Fetch the follower's remote actor to get their inbox URL. + remote, err := FetchActor(db, followerAPID) + if err != nil { + return fmt.Errorf("fetch follower actor: %w", err) + } + if remote.InboxURL == "" { + return fmt.Errorf("follower has no inbox URL") + } + + // Build Accept activity. + accept := map[string]any{ + "@context": "https://www.w3.org/ns/activitystreams", + "id": localActor.APID + "/activities/accept-" + fmt.Sprint(time.Now().UnixNano()), + "type": "Accept", + "actor": localActor.APID, + "object": follow, + } + + // Deliver asynchronously so inbox handler returns quickly. + go func() { + if err := DeliverActivity(localActor, accept, remote.InboxURL); err != nil { + log.Printf("federation: deliver Accept to %s: %v", remote.InboxURL, err) + return + } + // Store the outbound Accept. + db.Insert(&models.FederationActivity{ //nolint:errcheck + ActorAPID: localActor.APID, + Type: "Accept", + ObjectJSON: mustJSON(accept), + Direction: "outbound", + RemoteActor: followerAPID, + Published: time.Now().UTC(), + }) + log.Printf("federation: accepted Follow from %s", followerAPID) + }() + + return nil +} + +func handleAccept(db *xorm.Engine, localActor *models.FederationActor, activity map[string]any) { + // A remote actor accepted our Follow. Nothing to store beyond the inbox entry. + log.Printf("federation: received Accept for actor %s", localActor.APID) +} + +func handleUndo(db *xorm.Engine, localActor *models.FederationActor, activity map[string]any) { + // Common case: undo a Follow (unfollow). + obj, _ := activity["object"].(map[string]any) + if obj == nil { + return + } + if t, _ := obj["type"].(string); t == "Follow" { + log.Printf("federation: received Undo(Follow) for actor %s", localActor.APID) + } +} + +func mustJSON(v any) string { + b, _ := json.Marshal(v) + return string(b) +} diff --git a/internal/domain/federation/outbox.go b/internal/domain/federation/outbox.go new file mode 100644 index 0000000..c2045d0 --- /dev/null +++ b/internal/domain/federation/outbox.go @@ -0,0 +1,84 @@ +package federation + +import ( + "xorm.io/xorm" + + "github.com/forgeo/forgebucket/internal/models" +) + +const activitiesPerPage = 20 + +// Collection builds an ActivityStreams OrderedCollection (or page) for an actor's outbox. +// page=0 returns the collection summary; page≥1 returns a paginated OrderedCollectionPage. +func Collection(db *xorm.Engine, actorAPID string, outboxURL string, page int) map[string]any { + total, _ := db.Where("actor_ap_id = ? AND direction = 'outbound'", actorAPID). + Count(&models.FederationActivity{}) + + if page == 0 { + return map[string]any{ + "@context": "https://www.w3.org/ns/activitystreams", + "id": outboxURL, + "type": "OrderedCollection", + "totalItems": total, + "first": outboxURL + "?page=1", + } + } + + offset := (page - 1) * activitiesPerPage + var activities []models.FederationActivity + db.Where("actor_ap_id = ? AND direction = 'outbound'", actorAPID). + Desc("published"). + Limit(activitiesPerPage, offset). + Find(&activities) + + items := make([]any, 0, len(activities)) + for _, a := range activities { + items = append(items, rawJSON(a.ObjectJSON)) + } + + coll := map[string]any{ + "@context": "https://www.w3.org/ns/activitystreams", + "id": outboxURL + "?page=" + itoa(page), + "type": "OrderedCollectionPage", + "partOf": outboxURL, + "orderedItems": items, + } + if int64(offset+activitiesPerPage) < total { + coll["next"] = outboxURL + "?page=" + itoa(page+1) + } + return coll +} + +// StubCollection returns a minimal OrderedCollection with zero items. +// Used for followers/following until the social graph is implemented. +func StubCollection(collectionURL string) map[string]any { + return map[string]any{ + "@context": "https://www.w3.org/ns/activitystreams", + "id": collectionURL, + "type": "OrderedCollection", + "totalItems": 0, + "orderedItems": []any{}, + } +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + result := "" + for n > 0 { + result = string(rune('0'+n%10)) + result + n /= 10 + } + return result +} + +// rawJSON wraps a JSON string so it marshals as-is (not double-encoded). +type rawJSON string + +func (r rawJSON) MarshalJSON() ([]byte, error) { + if r == "" { + return []byte("null"), nil + } + return []byte(r), nil +} diff --git a/internal/domain/federation/remote.go b/internal/domain/federation/remote.go new file mode 100644 index 0000000..91151e4 --- /dev/null +++ b/internal/domain/federation/remote.go @@ -0,0 +1,121 @@ +package federation + +import ( + "encoding/json" + "fmt" + "net/http" + "time" + + "xorm.io/xorm" + + "github.com/forgeo/forgebucket/internal/models" +) + +const activityJSONType = "application/activity+json" + +// FetchActor retrieves a remote actor document by APID. If it is already cached +// in the remote_actor table it is returned immediately; otherwise it is fetched +// over HTTP and persisted before returning. +func FetchActor(db *xorm.Engine, apid string) (*models.RemoteActor, error) { + var cached models.RemoteActor + if found, _ := db.Where("ap_id = ?", apid).Get(&cached); found { + return &cached, nil + } + + client := &http.Client{Timeout: 10 * time.Second} + req, err := http.NewRequest("GET", apid, nil) + if err != nil { + return nil, fmt.Errorf("build request: %w", err) + } + req.Header.Set("Accept", activityJSONType+", application/ld+json") + + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("fetch actor %s: %w", apid, err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("fetch actor %s: HTTP %d", apid, resp.StatusCode) + } + + var doc map[string]any + if err := json.NewDecoder(resp.Body).Decode(&doc); err != nil { + return nil, fmt.Errorf("decode actor: %w", err) + } + + inbox, _ := doc["inbox"].(string) + pubKey := extractPublicKeyPEM(doc) + + actor := &models.RemoteActor{ + APID: apid, + InboxURL: inbox, + PublicKey: pubKey, + FetchedAt: time.Now().UTC(), + } + // Upsert: ignore duplicate key errors (concurrent fetch). + db.Insert(actor) //nolint:errcheck + + // Reload to get the DB-assigned ID. + db.Where("ap_id = ?", apid).Get(actor) //nolint:errcheck + return actor, nil +} + +// DeliverActivity POSTs a signed activity to a remote actor's inbox. +func DeliverActivity(localActor *models.FederationActor, activity map[string]any, recipientInbox string) error { + body, err := json.Marshal(activity) + if err != nil { + return fmt.Errorf("marshal activity: %w", err) + } + + req, err := http.NewRequest("POST", recipientInbox, jsonReader(body)) + if err != nil { + return fmt.Errorf("build request: %w", err) + } + req.Header.Set("Content-Type", activityJSONType) + req.Header.Set("Accept", activityJSONType) + + if err := Sign(req, localActor.APID+"#main-key", localActor.PrivateKey); err != nil { + return fmt.Errorf("sign: %w", err) + } + + client := &http.Client{Timeout: 15 * time.Second} + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("deliver to %s: %w", recipientInbox, err) + } + defer resp.Body.Close() + + if resp.StatusCode >= 400 { + return fmt.Errorf("deliver to %s: HTTP %d", recipientInbox, resp.StatusCode) + } + return nil +} + +func extractPublicKeyPEM(doc map[string]any) string { + pk, ok := doc["publicKey"].(map[string]any) + if !ok { + return "" + } + pem, _ := pk["publicKeyPem"].(string) + return pem +} + +// jsonReader wraps a byte slice in a reader that io.ReadCloser can use. +func jsonReader(data []byte) *bytesReader { + return &bytesReader{data: data, pos: 0} +} + +type bytesReader struct { + data []byte + pos int +} + +func (r *bytesReader) Read(p []byte) (int, error) { + if r.pos >= len(r.data) { + return 0, fmt.Errorf("EOF") + } + n := copy(p, r.data[r.pos:]) + r.pos += n + return n, nil +} diff --git a/internal/domain/federation/signatures.go b/internal/domain/federation/signatures.go new file mode 100644 index 0000000..25762c6 --- /dev/null +++ b/internal/domain/federation/signatures.go @@ -0,0 +1,175 @@ +package federation + +import ( + "crypto" + "crypto/rand" + "crypto/rsa" + "crypto/sha256" + "crypto/x509" + "encoding/base64" + "encoding/pem" + "fmt" + "net/http" + "strings" + "time" + + "xorm.io/xorm" +) + +// Sign adds an HTTP Signature header to req using the RSA private key. +// Implements draft-cavage-http-signatures (the fediverse de-facto standard). +// Signs: (request-target), host, date. If body is set, also signs digest. +func Sign(req *http.Request, keyID, privateKeyPEM string) error { + if req.Header.Get("Date") == "" { + req.Header.Set("Date", time.Now().UTC().Format(http.TimeFormat)) + } + if req.Header.Get("Host") == "" { + req.Header.Set("Host", req.URL.Host) + } + + method := strings.ToLower(req.Method) + target := req.URL.RequestURI() + signingString := fmt.Sprintf("(request-target): %s %s\nhost: %s\ndate: %s", + method, target, + req.Header.Get("Host"), + req.Header.Get("Date"), + ) + headers := "(request-target) host date" + + priv, err := parsePrivateKey(privateKeyPEM) + if err != nil { + return fmt.Errorf("parse private key: %w", err) + } + + h := sha256.Sum256([]byte(signingString)) + sig, err := rsa.SignPKCS1v15(rand.Reader, priv, crypto.SHA256, h[:]) + if err != nil { + return fmt.Errorf("sign: %w", err) + } + + req.Header.Set("Signature", fmt.Sprintf( + `keyId="%s",algorithm="rsa-sha256",headers="%s",signature="%s"`, + keyID, headers, base64.StdEncoding.EncodeToString(sig), + )) + return nil +} + +// Verify validates the HTTP Signature header on an incoming request. +// It fetches the sender's public key from their actor document (or the local DB). +func Verify(r *http.Request, db *xorm.Engine, instanceURL string) error { + sigHeader := r.Header.Get("Signature") + if sigHeader == "" { + return fmt.Errorf("missing Signature header") + } + + params := parseSignatureHeader(sigHeader) + keyID := params["keyId"] + sigB64 := params["signature"] + headersList := params["headers"] + if keyID == "" || sigB64 == "" { + return fmt.Errorf("malformed Signature header") + } + + sig, err := base64.StdEncoding.DecodeString(sigB64) + if err != nil { + return fmt.Errorf("decode signature: %w", err) + } + + // Fetch the public key for this keyId. + // keyId is typically "{actorURL}#main-key" — strip the fragment to get the actor APID. + actorAPID := strings.SplitN(keyID, "#", 2)[0] + pubKeyPEM, err := resolvePublicKey(db, actorAPID, instanceURL) + if err != nil { + return fmt.Errorf("resolve public key for %s: %w", actorAPID, err) + } + + // Reconstruct the signing string from the request. + signedHeaders := strings.Fields(headersList) + if len(signedHeaders) == 0 { + signedHeaders = []string{"date"} + } + var parts []string + for _, h := range signedHeaders { + switch h { + case "(request-target)": + parts = append(parts, fmt.Sprintf("(request-target): %s %s", + strings.ToLower(r.Method), r.URL.RequestURI())) + default: + parts = append(parts, h+": "+r.Header.Get(http.CanonicalHeaderKey(h))) + } + } + signingString := strings.Join(parts, "\n") + + pub, err := parsePublicKey(pubKeyPEM) + if err != nil { + return fmt.Errorf("parse public key: %w", err) + } + + h := sha256.Sum256([]byte(signingString)) + if err := rsa.VerifyPKCS1v15(pub, crypto.SHA256, h[:], sig); err != nil { + return fmt.Errorf("signature verification failed: %w", err) + } + return nil +} + +// ── helpers ────────────────────────────────────────────────────────────────── + +func parseSignatureHeader(header string) map[string]string { + params := make(map[string]string) + for _, part := range strings.Split(header, ",") { + part = strings.TrimSpace(part) + idx := strings.Index(part, "=") + if idx < 0 { + continue + } + key := strings.TrimSpace(part[:idx]) + val := strings.Trim(strings.TrimSpace(part[idx+1:]), `"`) + params[key] = val + } + return params +} + +func parsePrivateKey(pemStr string) (*rsa.PrivateKey, error) { + block, _ := pem.Decode([]byte(pemStr)) + if block == nil { + return nil, fmt.Errorf("no PEM block found") + } + return x509.ParsePKCS1PrivateKey(block.Bytes) +} + +func parsePublicKey(pemStr string) (*rsa.PublicKey, error) { + block, _ := pem.Decode([]byte(pemStr)) + if block == nil { + return nil, fmt.Errorf("no PEM block found") + } + pub, err := x509.ParsePKIXPublicKey(block.Bytes) + if err != nil { + return nil, err + } + rsaPub, ok := pub.(*rsa.PublicKey) + if !ok { + return nil, fmt.Errorf("not an RSA public key") + } + return rsaPub, nil +} + +// resolvePublicKey returns the public key PEM for an actor APID. +// Checks local actors first, then remote cache, then fetches from network. +func resolvePublicKey(db *xorm.Engine, actorAPID, instanceURL string) (string, error) { + // Check if it's a local actor. + var local struct { + PublicKey string `xorm:"public_key"` + } + if found, _ := db.Table("federation_actor"). + Where("ap_id = ?", actorAPID). + Cols("public_key").Get(&local); found && local.PublicKey != "" { + return local.PublicKey, nil + } + + // Fetch (and cache) from network. + remote, err := FetchActor(db, actorAPID) + if err != nil { + return "", err + } + return remote.PublicKey, nil +} diff --git a/internal/models/federation.go b/internal/models/federation.go index a0c8b44..8ade053 100644 --- a/internal/models/federation.go +++ b/internal/models/federation.go @@ -2,6 +2,28 @@ package models import "time" +// FederationActivity stores all inbound and outbound ActivityPub activities. +type FederationActivity struct { + ID int64 `xorm:"'id' pk autoincr" json:"id"` + ActorAPID string `xorm:"'actor_ap_id' notnull index varchar(500)" json:"actorApId"` + Type string `xorm:"'type' notnull varchar(50)" json:"type"` + ObjectJSON string `xorm:"'object_json' text" json:"objectJson"` + Direction string `xorm:"'direction' notnull varchar(10)" json:"direction"` // inbound|outbound + RemoteActor string `xorm:"'remote_actor' varchar(500)" json:"remoteActor"` + Published time.Time `xorm:"'published' index" json:"published"` + CreatedAt time.Time `xorm:"'created_at' created" json:"createdAt"` +} + +// RemoteActor caches public actor documents fetched from remote instances. +type RemoteActor struct { + ID int64 `xorm:"'id' pk autoincr"` + APID string `xorm:"'ap_id' notnull unique varchar(500)"` + InboxURL string `xorm:"'inbox_url' varchar(500)"` + PublicKey string `xorm:"'public_key' text"` + FetchedAt time.Time `xorm:"'fetched_at'"` + CreatedAt time.Time `xorm:"'created_at' created"` +} + type FederationActor struct { ID int64 `xorm:"'id' pk autoincr" json:"id"` UserID int64 `xorm:"'user_id' notnull unique index" json:"userId"` diff --git a/internal/models/migrations/001_init.go b/internal/models/migrations/001_init.go index 72215da..1a5f5d6 100644 --- a/internal/models/migrations/001_init.go +++ b/internal/models/migrations/001_init.go @@ -49,5 +49,8 @@ func Run(engine *xorm.Engine) error { if err := Run012(engine); err != nil { return err } - return Run013(engine) + if err := Run013(engine); err != nil { + return err + } + return Run014(engine) } diff --git a/internal/models/migrations/014_federation.go b/internal/models/migrations/014_federation.go new file mode 100644 index 0000000..c6fd326 --- /dev/null +++ b/internal/models/migrations/014_federation.go @@ -0,0 +1,13 @@ +package migrations + +import ( + "github.com/forgeo/forgebucket/internal/models" + "xorm.io/xorm" +) + +func Run014(engine *xorm.Engine) error { + return engine.Sync2( + &models.FederationActivity{}, + &models.RemoteActor{}, + ) +}