Add the runnable Gitea, PostgreSQL, and Hop control-plane foundation
Hop-State: A_06FN3QYA1N56NQYGZ81DWGR Hop-Proposal: R_06FN3QXJRYPPEQZQA6S69QG Hop-Task: T_06FN3MVGY3MT82ESQ89BND0 Hop-Attempt: AT_06FN3MVGY092BFA5MR9C7EG
This commit is contained in:
@@ -0,0 +1,50 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
HTTPAddr string
|
||||
DatabaseURL string
|
||||
AdminToken string
|
||||
GiteaWebhookSecret string
|
||||
GiteaBaseURL string
|
||||
GiteaAPIToken string
|
||||
}
|
||||
|
||||
func Load() (Config, error) {
|
||||
cfg := Config{
|
||||
HTTPAddr: value("HOP_HTTP_ADDR", ":8080"),
|
||||
DatabaseURL: os.Getenv("HOP_DATABASE_URL"),
|
||||
AdminToken: os.Getenv("HOP_ADMIN_TOKEN"),
|
||||
GiteaWebhookSecret: os.Getenv("HOP_GITEA_WEBHOOK_SECRET"),
|
||||
GiteaBaseURL: value("GITEA_BASE_URL", "http://localhost:3000"),
|
||||
GiteaAPIToken: os.Getenv("GITEA_API_TOKEN"),
|
||||
}
|
||||
|
||||
if cfg.DatabaseURL == "" {
|
||||
return Config{}, errors.New("HOP_DATABASE_URL is required")
|
||||
}
|
||||
if cfg.AdminToken == "" {
|
||||
return Config{}, errors.New("HOP_ADMIN_TOKEN is required")
|
||||
}
|
||||
if len(cfg.AdminToken) < 16 {
|
||||
return Config{}, errors.New("HOP_ADMIN_TOKEN must be at least 16 characters")
|
||||
}
|
||||
if cfg.GiteaWebhookSecret == "" {
|
||||
return Config{}, errors.New("HOP_GITEA_WEBHOOK_SECRET is required")
|
||||
}
|
||||
if len(cfg.GiteaWebhookSecret) < 16 {
|
||||
return Config{}, errors.New("HOP_GITEA_WEBHOOK_SECRET must be at least 16 characters")
|
||||
}
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
func value(name, fallback string) string {
|
||||
if v := os.Getenv(name); v != "" {
|
||||
return v
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
package config
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestLoadRequiresSecretsAndDatabase(t *testing.T) {
|
||||
t.Setenv("HOP_DATABASE_URL", "")
|
||||
t.Setenv("HOP_ADMIN_TOKEN", "")
|
||||
t.Setenv("HOP_GITEA_WEBHOOK_SECRET", "")
|
||||
if _, err := Load(); err == nil {
|
||||
t.Fatal("expected missing configuration error")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadDefaults(t *testing.T) {
|
||||
t.Setenv("HOP_DATABASE_URL", "postgres://example")
|
||||
t.Setenv("HOP_ADMIN_TOKEN", "admin-token-long-enough")
|
||||
t.Setenv("HOP_GITEA_WEBHOOK_SECRET", "webhook-secret-long-enough")
|
||||
t.Setenv("HOP_HTTP_ADDR", "")
|
||||
t.Setenv("GITEA_BASE_URL", "")
|
||||
cfg, err := Load()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cfg.HTTPAddr != ":8080" || cfg.GiteaBaseURL != "http://localhost:3000" {
|
||||
t.Fatalf("unexpected defaults: %#v", cfg)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
package gitea
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
var ErrTokenNotConfigured = errors.New("gitea API token is not configured")
|
||||
|
||||
type Repository struct {
|
||||
ID int64 `json:"id"`
|
||||
Owner Owner `json:"owner"`
|
||||
Name string `json:"name"`
|
||||
FullName string `json:"full_name"`
|
||||
CloneURL string `json:"clone_url"`
|
||||
DefaultBranch string `json:"default_branch"`
|
||||
}
|
||||
|
||||
type Owner struct {
|
||||
Login string `json:"login"`
|
||||
Username string `json:"username"`
|
||||
}
|
||||
|
||||
func (r Repository) OwnerName() string {
|
||||
if r.Owner.Login != "" {
|
||||
return r.Owner.Login
|
||||
}
|
||||
return r.Owner.Username
|
||||
}
|
||||
|
||||
type Client struct {
|
||||
baseURL *url.URL
|
||||
token string
|
||||
http *http.Client
|
||||
}
|
||||
|
||||
func NewClient(baseURL, token string) (*Client, error) {
|
||||
u, err := url.Parse(strings.TrimRight(baseURL, "/"))
|
||||
if err != nil || u.Scheme == "" || u.Host == "" {
|
||||
return nil, fmt.Errorf("invalid GITEA_BASE_URL %q", baseURL)
|
||||
}
|
||||
return &Client{baseURL: u, token: token, http: &http.Client{Timeout: 10 * time.Second}}, nil
|
||||
}
|
||||
|
||||
func (c *Client) Repository(ctx context.Context, owner, name string) (Repository, error) {
|
||||
if c.token == "" {
|
||||
return Repository{}, ErrTokenNotConfigured
|
||||
}
|
||||
u := c.baseURL.JoinPath("api", "v1", "repos", owner, name)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
|
||||
if err != nil {
|
||||
return Repository{}, err
|
||||
}
|
||||
req.Header.Set("Authorization", "token "+c.token)
|
||||
req.Header.Set("Accept", "application/json")
|
||||
|
||||
resp, err := c.http.Do(req)
|
||||
if err != nil {
|
||||
return Repository{}, fmt.Errorf("request repository: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return Repository{}, fmt.Errorf("gitea repository lookup returned %s", resp.Status)
|
||||
}
|
||||
|
||||
var repo Repository
|
||||
if err := json.NewDecoder(resp.Body).Decode(&repo); err != nil {
|
||||
return Repository{}, fmt.Errorf("decode repository: %w", err)
|
||||
}
|
||||
if repo.ID <= 0 || repo.OwnerName() == "" || repo.Name == "" {
|
||||
return Repository{}, errors.New("gitea returned an incomplete repository")
|
||||
}
|
||||
return repo, nil
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package gitea
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestRepositoryUsesTokenAndDecodesResponse(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/gitea/api/v1/repos/hop/demo" {
|
||||
t.Fatalf("path = %q", r.URL.Path)
|
||||
}
|
||||
if got := r.Header.Get("Authorization"); got != "token gitea-token" {
|
||||
t.Fatalf("authorization = %q", got)
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"id":42,"owner":{"login":"hop"},"name":"demo","full_name":"hop/demo","clone_url":"http://gitea/hop/demo.git","default_branch":"main"}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
client, err := NewClient(server.URL+"/gitea", "gitea-token")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
repository, err := client.Repository(context.Background(), "hop", "demo")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if repository.ID != 42 || repository.OwnerName() != "hop" || repository.DefaultBranch != "main" {
|
||||
t.Fatalf("unexpected repository: %#v", repository)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepositoryRequiresToken(t *testing.T) {
|
||||
client, err := NewClient("http://gitea.test", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := client.Repository(context.Background(), "hop", "demo"); err != ErrTokenNotConfigured {
|
||||
t.Fatalf("error = %v, want %v", err, ErrTokenNotConfigured)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package gitea
|
||||
|
||||
import (
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
)
|
||||
|
||||
func VerifySignature(body []byte, signature, secret string) bool {
|
||||
provided, err := hex.DecodeString(signature)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
mac := hmac.New(sha256.New, []byte(secret))
|
||||
_, _ = mac.Write(body)
|
||||
return hmac.Equal(provided, mac.Sum(nil))
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package gitea
|
||||
|
||||
import (
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestVerifySignature(t *testing.T) {
|
||||
body := []byte(`{"hello":"hop"}`)
|
||||
mac := hmac.New(sha256.New, []byte("secret"))
|
||||
_, _ = mac.Write(body)
|
||||
signature := hex.EncodeToString(mac.Sum(nil))
|
||||
|
||||
if !VerifySignature(body, signature, "secret") {
|
||||
t.Fatal("expected valid signature")
|
||||
}
|
||||
if VerifySignature(body, signature, "wrong") {
|
||||
t.Fatal("accepted signature with wrong secret")
|
||||
}
|
||||
if VerifySignature(body, "not-hex", "secret") {
|
||||
t.Fatal("accepted malformed signature")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"crypto/subtle"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"hopweb/internal/gitea"
|
||||
"hopweb/internal/store"
|
||||
)
|
||||
|
||||
const maxWebhookBody = 2 << 20
|
||||
|
||||
type giteaClient interface {
|
||||
Repository(context.Context, string, string) (gitea.Repository, error)
|
||||
}
|
||||
|
||||
type server struct {
|
||||
store store.Store
|
||||
gitea giteaClient
|
||||
adminToken string
|
||||
webhookSecret string
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
func New(data store.Store, client giteaClient, adminToken, webhookSecret string, logger *slog.Logger) http.Handler {
|
||||
s := &server{store: data, gitea: client, adminToken: adminToken, webhookSecret: webhookSecret, logger: logger}
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("GET /healthz", s.health)
|
||||
mux.HandleFunc("GET /readyz", s.ready)
|
||||
mux.Handle("GET /api/v1/repositories", s.requireAdmin(http.HandlerFunc(s.listRepositories)))
|
||||
mux.Handle("POST /api/v1/repositories/link", s.requireAdmin(http.HandlerFunc(s.linkRepository)))
|
||||
mux.HandleFunc("POST /api/v1/gitea/webhooks", s.giteaWebhook)
|
||||
return requestLogger(logger, mux)
|
||||
}
|
||||
|
||||
func (s *server) health(w http.ResponseWriter, _ *http.Request) {
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
||||
}
|
||||
|
||||
func (s *server) ready(w http.ResponseWriter, r *http.Request) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
|
||||
defer cancel()
|
||||
if err := s.store.Ping(ctx); err != nil {
|
||||
writeError(w, http.StatusServiceUnavailable, "database unavailable")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ready"})
|
||||
}
|
||||
|
||||
func (s *server) listRepositories(w http.ResponseWriter, r *http.Request) {
|
||||
repositories, err := s.store.ListRepositories(r.Context())
|
||||
if err != nil {
|
||||
s.logger.Error("list repositories", "error", err)
|
||||
writeError(w, http.StatusInternalServerError, "could not list repositories")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{"repositories": repositories})
|
||||
}
|
||||
|
||||
func (s *server) linkRepository(w http.ResponseWriter, r *http.Request) {
|
||||
var input struct {
|
||||
Owner string `json:"owner"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
if err := decodeJSON(w, r, &input); err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
if !validSlug(input.Owner) || !validSlug(input.Name) {
|
||||
writeError(w, http.StatusBadRequest, "owner and name are required and cannot contain slashes")
|
||||
return
|
||||
}
|
||||
repository, err := s.gitea.Repository(r.Context(), input.Owner, input.Name)
|
||||
if err != nil {
|
||||
if errors.Is(err, gitea.ErrTokenNotConfigured) {
|
||||
writeError(w, http.StatusServiceUnavailable, err.Error())
|
||||
return
|
||||
}
|
||||
s.logger.Warn("gitea repository lookup failed", "owner", input.Owner, "name", input.Name, "error", err)
|
||||
writeError(w, http.StatusBadGateway, "could not load repository from Gitea")
|
||||
return
|
||||
}
|
||||
saved, err := s.store.UpsertRepository(r.Context(), repository)
|
||||
if err != nil {
|
||||
s.logger.Error("link repository", "error", err)
|
||||
writeError(w, http.StatusInternalServerError, "could not link repository")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusCreated, saved)
|
||||
}
|
||||
|
||||
func (s *server) giteaWebhook(w http.ResponseWriter, r *http.Request) {
|
||||
body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxWebhookBody))
|
||||
if err != nil {
|
||||
writeError(w, http.StatusRequestEntityTooLarge, "webhook payload too large")
|
||||
return
|
||||
}
|
||||
if !gitea.VerifySignature(body, r.Header.Get("X-Gitea-Signature"), s.webhookSecret) {
|
||||
writeError(w, http.StatusUnauthorized, "invalid webhook signature")
|
||||
return
|
||||
}
|
||||
deliveryID := strings.TrimSpace(r.Header.Get("X-Gitea-Delivery"))
|
||||
event := strings.TrimSpace(r.Header.Get("X-Gitea-Event"))
|
||||
if deliveryID == "" || event == "" {
|
||||
writeError(w, http.StatusBadRequest, "missing Gitea delivery or event header")
|
||||
return
|
||||
}
|
||||
var payload struct {
|
||||
Repository *gitea.Repository `json:"repository"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &payload); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid JSON payload")
|
||||
return
|
||||
}
|
||||
digest := sha256.Sum256(body)
|
||||
duplicate, err := s.store.RecordWebhook(r.Context(), store.WebhookDelivery{
|
||||
DeliveryID: deliveryID, Event: event, EventType: r.Header.Get("X-Gitea-Event-Type"),
|
||||
PayloadSHA256: hex.EncodeToString(digest[:]), Repository: payload.Repository,
|
||||
})
|
||||
if err != nil {
|
||||
s.logger.Error("record webhook", "delivery_id", deliveryID, "error", err)
|
||||
writeError(w, http.StatusInternalServerError, "could not record webhook")
|
||||
return
|
||||
}
|
||||
if duplicate {
|
||||
w.Header().Set("X-Hop-Duplicate", "true")
|
||||
}
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
func (s *server) requireAdmin(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
authorization := r.Header.Get("Authorization")
|
||||
if !strings.HasPrefix(authorization, "Bearer ") {
|
||||
writeError(w, http.StatusUnauthorized, "unauthorized")
|
||||
return
|
||||
}
|
||||
provided := strings.TrimPrefix(authorization, "Bearer ")
|
||||
if len(provided) != len(s.adminToken) || subtle.ConstantTimeCompare([]byte(provided), []byte(s.adminToken)) != 1 {
|
||||
writeError(w, http.StatusUnauthorized, "unauthorized")
|
||||
return
|
||||
}
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
|
||||
func validSlug(value string) bool {
|
||||
value = strings.TrimSpace(value)
|
||||
return value != "" && !strings.ContainsAny(value, "/\\")
|
||||
}
|
||||
|
||||
func decodeJSON(w http.ResponseWriter, r *http.Request, target any) error {
|
||||
r.Body = http.MaxBytesReader(w, r.Body, 64<<10)
|
||||
decoder := json.NewDecoder(r.Body)
|
||||
decoder.DisallowUnknownFields()
|
||||
if err := decoder.Decode(target); err != nil {
|
||||
return errors.New("invalid JSON body")
|
||||
}
|
||||
if err := decoder.Decode(&struct{}{}); err != io.EOF {
|
||||
return errors.New("request body must contain one JSON object")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func writeError(w http.ResponseWriter, status int, message string) {
|
||||
writeJSON(w, status, map[string]any{"error": map[string]string{"message": message}})
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, body any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(body)
|
||||
}
|
||||
|
||||
func requestLogger(logger *slog.Logger, next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
started := time.Now()
|
||||
next.ServeHTTP(w, r)
|
||||
logger.Info("request", "method", r.Method, "path", r.URL.Path, "duration_ms", time.Since(started).Milliseconds())
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"hopweb/internal/gitea"
|
||||
"hopweb/internal/store"
|
||||
)
|
||||
|
||||
type fakeStore struct {
|
||||
repositories []store.Repository
|
||||
delivery store.WebhookDelivery
|
||||
duplicate bool
|
||||
}
|
||||
|
||||
func (f *fakeStore) Ping(context.Context) error { return nil }
|
||||
func (f *fakeStore) ListRepositories(context.Context) ([]store.Repository, error) {
|
||||
return f.repositories, nil
|
||||
}
|
||||
func (f *fakeStore) UpsertRepository(_ context.Context, repository gitea.Repository) (store.Repository, error) {
|
||||
saved := store.Repository{ID: "RP_TEST", GiteaID: repository.ID, Owner: repository.OwnerName(), Name: repository.Name, FullName: repository.FullName}
|
||||
f.repositories = append(f.repositories, saved)
|
||||
return saved, nil
|
||||
}
|
||||
func (f *fakeStore) RecordWebhook(_ context.Context, delivery store.WebhookDelivery) (bool, error) {
|
||||
f.delivery = delivery
|
||||
return f.duplicate, nil
|
||||
}
|
||||
|
||||
type fakeGitea struct {
|
||||
repository gitea.Repository
|
||||
err error
|
||||
}
|
||||
|
||||
func (f fakeGitea) Repository(context.Context, string, string) (gitea.Repository, error) {
|
||||
return f.repository, f.err
|
||||
}
|
||||
|
||||
func testServer(data store.Store, client giteaClient) http.Handler {
|
||||
return New(data, client, "admin-secret", "webhook-secret", slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
}
|
||||
|
||||
func TestAdminAuthentication(t *testing.T) {
|
||||
handler := testServer(&fakeStore{}, fakeGitea{})
|
||||
for _, header := range []string{"", "admin-secret", "Bearer wrong"} {
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/repositories", nil)
|
||||
req.Header.Set("Authorization", header)
|
||||
response := httptest.NewRecorder()
|
||||
handler.ServeHTTP(response, req)
|
||||
if response.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("header %q: status = %d, want %d", header, response.Code, http.StatusUnauthorized)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestLinkRepository(t *testing.T) {
|
||||
data := &fakeStore{}
|
||||
client := fakeGitea{repository: gitea.Repository{ID: 42, Owner: gitea.Owner{Login: "hop"}, Name: "demo", FullName: "hop/demo"}}
|
||||
handler := testServer(data, client)
|
||||
req := httptest.NewRequest(http.MethodPost, "/api/v1/repositories/link", bytes.NewBufferString(`{"owner":"hop","name":"demo"}`))
|
||||
req.Header.Set("Authorization", "Bearer admin-secret")
|
||||
response := httptest.NewRecorder()
|
||||
handler.ServeHTTP(response, req)
|
||||
if response.Code != http.StatusCreated {
|
||||
t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusCreated, response.Body.String())
|
||||
}
|
||||
if len(data.repositories) != 1 || data.repositories[0].GiteaID != 42 {
|
||||
t.Fatalf("repository not linked: %#v", data.repositories)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhookVerificationAndDeduplication(t *testing.T) {
|
||||
data := &fakeStore{duplicate: true}
|
||||
handler := testServer(data, fakeGitea{err: errors.New("unused")})
|
||||
body := []byte(`{"repository":{"id":42,"owner":{"login":"hop"},"name":"demo","full_name":"hop/demo"}}`)
|
||||
mac := hmac.New(sha256.New, []byte("webhook-secret"))
|
||||
_, _ = mac.Write(body)
|
||||
|
||||
req := httptest.NewRequest(http.MethodPost, "/api/v1/gitea/webhooks", bytes.NewReader(body))
|
||||
req.Header.Set("X-Gitea-Signature", hex.EncodeToString(mac.Sum(nil)))
|
||||
req.Header.Set("X-Gitea-Delivery", "delivery-1")
|
||||
req.Header.Set("X-Gitea-Event", "push")
|
||||
response := httptest.NewRecorder()
|
||||
handler.ServeHTTP(response, req)
|
||||
|
||||
if response.Code != http.StatusNoContent {
|
||||
t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusNoContent, response.Body.String())
|
||||
}
|
||||
if response.Header().Get("X-Hop-Duplicate") != "true" {
|
||||
t.Fatal("expected duplicate response header")
|
||||
}
|
||||
if data.delivery.DeliveryID != "delivery-1" || data.delivery.Repository == nil || data.delivery.Repository.ID != 42 {
|
||||
t.Fatalf("delivery not recorded: %#v", data.delivery)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhookRejectsInvalidSignature(t *testing.T) {
|
||||
handler := testServer(&fakeStore{}, fakeGitea{})
|
||||
req := httptest.NewRequest(http.MethodPost, "/api/v1/gitea/webhooks", bytes.NewBufferString(`{}`))
|
||||
req.Header.Set("X-Gitea-Signature", "bad")
|
||||
response := httptest.NewRecorder()
|
||||
handler.ServeHTTP(response, req)
|
||||
if response.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("status = %d, want %d", response.Code, http.StatusUnauthorized)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
package ids
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"encoding/base32"
|
||||
"time"
|
||||
)
|
||||
|
||||
var encoding = base32.NewEncoding("0123456789ABCDEFGHJKMNPQRSTVWXYZ").WithPadding(base32.NoPadding)
|
||||
|
||||
func Repository() (string, error) {
|
||||
value := make([]byte, 16)
|
||||
milliseconds := uint64(time.Now().UnixMilli())
|
||||
value[0] = byte(milliseconds >> 40)
|
||||
value[1] = byte(milliseconds >> 32)
|
||||
value[2] = byte(milliseconds >> 24)
|
||||
value[3] = byte(milliseconds >> 16)
|
||||
value[4] = byte(milliseconds >> 8)
|
||||
value[5] = byte(milliseconds)
|
||||
if _, err := rand.Read(value[6:]); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return "RP_" + encoding.EncodeToString(value), nil
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
package ids
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestRepository(t *testing.T) {
|
||||
first, err := Repository()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
second, err := Repository()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.HasPrefix(first, "RP_") || len(first) != 29 {
|
||||
t.Fatalf("unexpected repository ID %q", first)
|
||||
}
|
||||
if first == second {
|
||||
t.Fatal("generated duplicate IDs")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
CREATE TABLE repositories (
|
||||
id text PRIMARY KEY,
|
||||
gitea_id bigint NOT NULL UNIQUE CHECK (gitea_id > 0),
|
||||
owner text NOT NULL,
|
||||
name text NOT NULL,
|
||||
full_name text NOT NULL,
|
||||
clone_url text NOT NULL DEFAULT '',
|
||||
default_branch text NOT NULL DEFAULT '',
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
updated_at timestamptz NOT NULL DEFAULT now(),
|
||||
UNIQUE (owner, name)
|
||||
);
|
||||
|
||||
CREATE TABLE webhook_deliveries (
|
||||
delivery_id text PRIMARY KEY,
|
||||
event text NOT NULL,
|
||||
event_type text NOT NULL DEFAULT '',
|
||||
payload_sha256 text NOT NULL,
|
||||
received_at timestamptz NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"embed"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"hopweb/internal/gitea"
|
||||
"hopweb/internal/ids"
|
||||
)
|
||||
|
||||
//go:embed migrations/*.sql
|
||||
var migrations embed.FS
|
||||
|
||||
type Repository struct {
|
||||
ID string `json:"id"`
|
||||
GiteaID int64 `json:"gitea_id"`
|
||||
Owner string `json:"owner"`
|
||||
Name string `json:"name"`
|
||||
FullName string `json:"full_name"`
|
||||
CloneURL string `json:"clone_url"`
|
||||
DefaultBranch string `json:"default_branch"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
type WebhookDelivery struct {
|
||||
DeliveryID string
|
||||
Event string
|
||||
EventType string
|
||||
PayloadSHA256 string
|
||||
Repository *gitea.Repository
|
||||
}
|
||||
|
||||
type Store interface {
|
||||
Ping(context.Context) error
|
||||
ListRepositories(context.Context) ([]Repository, error)
|
||||
UpsertRepository(context.Context, gitea.Repository) (Repository, error)
|
||||
RecordWebhook(context.Context, WebhookDelivery) (bool, error)
|
||||
}
|
||||
|
||||
type Postgres struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func Open(ctx context.Context, databaseURL string) (*Postgres, error) {
|
||||
pool, err := pgxpool.New(ctx, databaseURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := pool.Ping(ctx); err != nil {
|
||||
pool.Close()
|
||||
return nil, err
|
||||
}
|
||||
return &Postgres{pool: pool}, nil
|
||||
}
|
||||
|
||||
func (p *Postgres) Close() { p.pool.Close() }
|
||||
|
||||
func (p *Postgres) Ping(ctx context.Context) error { return p.pool.Ping(ctx) }
|
||||
|
||||
func (p *Postgres) Migrate(ctx context.Context) error {
|
||||
if _, err := p.pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`); err != nil {
|
||||
return err
|
||||
}
|
||||
entries, err := migrations.ReadDir("migrations")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })
|
||||
for _, entry := range entries {
|
||||
name := entry.Name()
|
||||
var exists bool
|
||||
if err := p.pool.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migrations WHERE name=$1)`, name).Scan(&exists); err != nil {
|
||||
return err
|
||||
}
|
||||
if exists {
|
||||
continue
|
||||
}
|
||||
sql, err := migrations.ReadFile("migrations/" + name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tx, err := p.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = tx.Exec(ctx, string(sql)); err == nil {
|
||||
_, err = tx.Exec(ctx, `INSERT INTO schema_migrations (name) VALUES ($1)`, name)
|
||||
}
|
||||
if err != nil {
|
||||
_ = tx.Rollback(ctx)
|
||||
return fmt.Errorf("apply migration %s: %w", name, err)
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Postgres) ListRepositories(ctx context.Context) ([]Repository, error) {
|
||||
rows, err := p.pool.Query(ctx, `SELECT id, gitea_id, owner, name, full_name, clone_url, default_branch, created_at, updated_at FROM repositories ORDER BY full_name`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
repositories := make([]Repository, 0)
|
||||
for rows.Next() {
|
||||
var repository Repository
|
||||
if err := rows.Scan(&repository.ID, &repository.GiteaID, &repository.Owner, &repository.Name, &repository.FullName, &repository.CloneURL, &repository.DefaultBranch, &repository.CreatedAt, &repository.UpdatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
repositories = append(repositories, repository)
|
||||
}
|
||||
return repositories, rows.Err()
|
||||
}
|
||||
|
||||
func (p *Postgres) UpsertRepository(ctx context.Context, repository gitea.Repository) (Repository, error) {
|
||||
id, err := ids.Repository()
|
||||
if err != nil {
|
||||
return Repository{}, err
|
||||
}
|
||||
return upsertRepository(ctx, p.pool, id, repository)
|
||||
}
|
||||
|
||||
type querier interface {
|
||||
QueryRow(context.Context, string, ...any) pgx.Row
|
||||
}
|
||||
|
||||
func upsertRepository(ctx context.Context, q querier, id string, repository gitea.Repository) (Repository, error) {
|
||||
owner := repository.OwnerName()
|
||||
fullName := repository.FullName
|
||||
if fullName == "" {
|
||||
fullName = owner + "/" + repository.Name
|
||||
}
|
||||
var saved Repository
|
||||
err := q.QueryRow(ctx, `
|
||||
INSERT INTO repositories (id, gitea_id, owner, name, full_name, clone_url, default_branch)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7)
|
||||
ON CONFLICT (gitea_id) DO UPDATE SET
|
||||
owner=EXCLUDED.owner, name=EXCLUDED.name, full_name=EXCLUDED.full_name,
|
||||
clone_url=EXCLUDED.clone_url, default_branch=EXCLUDED.default_branch, updated_at=now()
|
||||
RETURNING id, gitea_id, owner, name, full_name, clone_url, default_branch, created_at, updated_at`,
|
||||
id, repository.ID, owner, repository.Name, fullName, repository.CloneURL, repository.DefaultBranch,
|
||||
).Scan(&saved.ID, &saved.GiteaID, &saved.Owner, &saved.Name, &saved.FullName, &saved.CloneURL, &saved.DefaultBranch, &saved.CreatedAt, &saved.UpdatedAt)
|
||||
return saved, err
|
||||
}
|
||||
|
||||
func (p *Postgres) RecordWebhook(ctx context.Context, delivery WebhookDelivery) (bool, error) {
|
||||
tx, err := p.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
result, err := tx.Exec(ctx, `INSERT INTO webhook_deliveries (delivery_id, event, event_type, payload_sha256) VALUES ($1,$2,$3,$4) ON CONFLICT DO NOTHING`, delivery.DeliveryID, delivery.Event, delivery.EventType, delivery.PayloadSHA256)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if result.RowsAffected() == 0 {
|
||||
var existingEvent, existingEventType, existingDigest string
|
||||
if err := tx.QueryRow(ctx, `SELECT event, event_type, payload_sha256 FROM webhook_deliveries WHERE delivery_id=$1`, delivery.DeliveryID).Scan(&existingEvent, &existingEventType, &existingDigest); err != nil {
|
||||
return false, err
|
||||
}
|
||||
if existingEvent != delivery.Event || existingEventType != delivery.EventType || existingDigest != delivery.PayloadSHA256 {
|
||||
return false, fmt.Errorf("webhook delivery ID %q was reused with different content", delivery.DeliveryID)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
if delivery.Repository != nil && delivery.Repository.ID > 0 {
|
||||
id, err := ids.Repository()
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if _, err := upsertRepository(ctx, tx, id, *delivery.Repository); err != nil {
|
||||
return false, err
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return false, nil
|
||||
}
|
||||
|
||||
var _ Store = (*Postgres)(nil)
|
||||
Reference in New Issue
Block a user