Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566package api
import ( "context" "encoding/json" "errors" "fmt" "io" "log/slog" "net/http" "net/url" "strconv" "strings" "time" "unicode/utf8"
"github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/go-chi/chi/v5" "github.com/google/uuid" "tangled.org/core/api/org_tangled" "tangled.org/core/log" "tangled.org/core/migrator/config" "tangled.org/core/migrator/crypto" "tangled.org/core/migrator/db" "tangled.org/core/migrator/git" migratoroauth "tangled.org/core/migrator/oauth" xrpcerr "tangled.org/core/xrpc/errors" "tangled.org/core/xrpc/serviceauth")
const ( maxJobsPerTask = 50 maxRequestID = 128 // schema is 140 graphemes, rune count is an upper bound maxDescription = 140 defaultListLimit = 20 maxListLimit = 100)
type GrantChecker interface { HasSession(context.Context, string) bool}
type GrantProber interface { UsableGrant(ctx context.Context, ownerDid string) error}
type Server struct { db *db.DB cfg *config.Config signer *serviceauth.Signer serviceAuth *serviceauth.ServiceAuth logger *slog.Logger grants GrantChecker gitRunner git.CommandRunner proxyAddr string}
func NewServer( database *db.DB, cfg *config.Config, signer *serviceauth.Signer, serviceAuth *serviceauth.ServiceAuth, grants GrantChecker, logger *slog.Logger, gitRunner git.CommandRunner, proxyAddr string,) *Server { return &Server{ db: database, cfg: cfg, signer: signer, serviceAuth: serviceAuth, grants: grants, logger: log.SubLogger(logger, "api"), gitRunner: gitRunner, proxyAddr: proxyAddr, }}
func (s *Server) gitRunnerOrDefault() git.CommandRunner { if s.gitRunner != nil { return s.gitRunner } return git.RealCommandRunner{}}
func (s *Server) Routes() http.Handler { r := chi.NewRouter()
r.Get("/.well-known/did.json", s.handleDIDDoc)
r.Mount("/xrpc", s.xrpcRoutes())
return r}
func (s *Server) xrpcRoutes() http.Handler { r := chi.NewRouter()
r.Use(s.cors)
r.Use(s.serviceAuth.VerifyServiceAuth)
r.Post("/"+org_tangled.TempMigratorCreateTaskNSID, s.handleCreateTask) r.Get("/"+org_tangled.TempMigratorGetTaskNSID, s.handleGetTask) r.Get("/"+org_tangled.TempMigratorDescribeSourceNSID, s.handleDescribeSource) r.Get("/"+org_tangled.TempMigratorListTasksNSID, s.handleListTasks) r.Post("/"+org_tangled.TempMigratorRetryJobNSID, s.handleRetryJob)
return r}
func (s *Server) cors(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { h := w.Header() h.Set("Access-Control-Allow-Origin", "*") h.Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS") h.Set("Access-Control-Allow-Headers", "Content-Type, Authorization") h.Set("Access-Control-Max-Age", "86400")
if r.Method == http.MethodOptions { w.WriteHeader(http.StatusNoContent) return }
next.ServeHTTP(w, r) })}
func (s *Server) handleDIDDoc(w http.ResponseWriter, r *http.Request) { if s.signer == nil { s.writeError(w, internalError, http.StatusServiceUnavailable) return }
// knot2 resolver requires an atproto_pds entry on did:web issuers doc := s.signer.DIDDocument() doc.Service = []identity.DocService{{ ID: "#atproto_pds", Type: "AtprotoPersonalDataServer", ServiceEndpoint: s.cfg.ServiceURL(), }}
w.Header().Set("Content-Type", "application/did+json") _ = json.NewEncoder(w).Encode(doc)}
func (s *Server) handleCreateTask(w http.ResponseWriter, r *http.Request) { owner, ok := actorDid(r) if !ok { s.writeError(w, missingActorError, http.StatusUnauthorized) return }
var input org_tangled.TempMigratorCreateTask_Input if err := decodeStrictJSON(w, r, &input, 1<<20); err != nil { s.writeError(w, invalidRequest("invalid request body: "+err.Error()), http.StatusBadRequest) return }
if input.RequestId == "" { s.writeError(w, invalidRequest("requestId is required"), http.StatusBadRequest) return } if utf8.RuneCountInString(input.RequestId) > maxRequestID { s.writeError(w, invalidRequest(fmt.Sprintf("requestId must be %d characters or fewer", maxRequestID)), http.StatusBadRequest) return } if len(input.Jobs) == 0 { s.writeError(w, invalidRequest("at least one job is required"), http.StatusBadRequest) return } if len(input.Jobs) > maxJobsPerTask { s.writeError(w, invalidRequest(fmt.Sprintf("too many jobs: at most %d allowed, got %d", maxJobsPerTask, len(input.Jobs))), http.StatusBadRequest) return }
hasPrivate := false seenNames := make(map[string]struct{}, len(input.Jobs)) for i, job := range input.Jobs { if job == nil { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: missing", i)), http.StatusBadRequest) return } if err := ValidateRepoName(job.Name); err != nil { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: invalid name: %v", i, err)), http.StatusBadRequest) return } name := strings.ToLower(job.Name) if _, twice := seenNames[name]; twice { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: %q is asked for twice", i, job.Name)), http.StatusBadRequest) return } seenNames[name] = struct{}{} if description := DescriptionOf(job); utf8.RuneCountInString(description) > maxDescription { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: description must be %d characters or fewer", i, maxDescription)), http.StatusBadRequest) return } if err := ValidateKnotDID(job.KnotDid); err != nil { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: invalid knotDid: %v", i, err)), http.StatusBadRequest) return } if job.Private != nil && *job.Private { if err := ValidateGitHubSourceURL(job.SourceUrl, s.cfg.GitHubHost); err != nil { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: %v", i, err)), http.StatusBadRequest) return } hasPrivate = true } else if err := ValidateSourceURL(job.SourceUrl); err != nil { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: invalid sourceUrl: %v", i, err)), http.StatusBadRequest) return } }
if s.grants == nil || !s.grants.HasSession(r.Context(), owner) { s.writeError(w, grantRequired(s.cfg.ServiceURL(), owner), http.StatusPreconditionRequired) return } if prober, ok := s.grants.(GrantProber); ok { if err := prober.UsableGrant(r.Context(), owner); err != nil { if errors.Is(err, migratoroauth.ErrGrantRequired) { s.writeError(w, grantRequired(s.cfg.ServiceURL(), owner), http.StatusPreconditionRequired) return } // transient probe failures do not block creation; retried at job time s.logger.Error("grant probe failed", "ownerDid", owner, "err", err) } }
taskID := uuid.New().String()
var encryptedToken *string if hasPrivate { if input.GithubToken == nil || strings.TrimSpace(*input.GithubToken) == "" { s.writeError(w, invalidRequest("githubToken is required when one or more jobs are private"), http.StatusBadRequest) return } aad := crypto.ComputeAAD(owner, taskID, input.RequestId) enc, err := crypto.Encrypt(s.cfg.ParsedMasterKey, *input.GithubToken, aad) if err != nil { s.logger.Error("failed to encrypt githubToken", "err", err) s.writeError(w, internalError, http.StatusInternalServerError) return } encryptedToken = &enc }
dbJobs := mapSlice(input.Jobs, func(job *org_tangled.TempMigratorDefs_NewJob) db.CreateJobInput { return db.CreateJobInput{ Name: job.Name, KnotDid: job.KnotDid, SourceURL: job.SourceUrl, Description: DescriptionOf(job), Private: job.Private != nil && *job.Private, } })
batch, jobs, reused, err := s.db.CreateBatch(r.Context(), db.CreateBatchInput{ ID: taskID, OwnerDid: owner, RequestID: input.RequestId, RequestDigest: ComputeRequestDigest(input.Jobs), EncryptedToken: encryptedToken, Jobs: dbJobs, }) if errors.Is(err, db.ErrConflict) { s.writeError(w, taskConflict(), http.StatusConflict) return } if err != nil { s.logger.Error("failed to create task", "err", err, "ownerDid", owner) s.writeError(w, internalError, http.StatusInternalServerError) return }
status := http.StatusAccepted if reused { status = http.StatusOK }
s.writeTask(w, batch.Task(jobs), status)}
func (s *Server) handleGetTask(w http.ResponseWriter, r *http.Request) { owner, ok := actorDid(r) if !ok { s.writeError(w, missingActorError, http.StatusUnauthorized) return }
taskID := r.URL.Query().Get("taskId") if taskID == "" { s.writeError(w, invalidRequest("taskId is required"), http.StatusBadRequest) return }
batch, jobs, err := s.db.GetBatch(r.Context(), taskID) if errors.Is(err, db.ErrNotFound) || (err == nil && batch.OwnerDid != owner) { s.writeError(w, taskNotFound(), http.StatusNotFound) return } if err != nil { s.logger.Error("failed to read task", "err", err, "taskId", taskID) s.writeError(w, internalError, http.StatusInternalServerError) return }
s.writeTask(w, batch.Task(jobs), http.StatusOK)}
func (s *Server) handleListTasks(w http.ResponseWriter, r *http.Request) { owner, ok := actorDid(r) if !ok { s.writeError(w, missingActorError, http.StatusUnauthorized) return }
limit := defaultListLimit if raw := r.URL.Query().Get("limit"); raw != "" { parsed, err := strconv.Atoi(raw) if err != nil { s.writeError(w, invalidRequest("limit must be an integer"), http.StatusBadRequest) return } limit = min(max(parsed, 1), maxListLimit) }
tasks, err := s.db.GetRecentTasksForOwner(r.Context(), owner, limit) if err != nil { s.logger.Error("failed to list tasks", "err", err, "ownerDid", owner) s.writeError(w, internalError, http.StatusInternalServerError) return }
out := org_tangled.TempMigratorListTasks_Output{Tasks: mapSlice(tasks, taskView)} s.writeJSON(w, out, http.StatusOK)}
func (s *Server) handleRetryJob(w http.ResponseWriter, r *http.Request) { owner, ok := actorDid(r) if !ok { s.writeError(w, missingActorError, http.StatusUnauthorized) return }
var input org_tangled.TempMigratorRetryJob_Input if err := decodeStrictJSON(w, r, &input, 64*1024); err != nil { s.writeError(w, invalidRequest("invalid request body: "+err.Error()), http.StatusBadRequest) return } if input.TaskId == "" { s.writeError(w, invalidRequest("taskId is required"), http.StatusBadRequest) return }
jobID, err := strconv.ParseInt(input.JobId, 10, 64) if err != nil { s.writeError(w, invalidRequest("jobId must be the id createTask returned for the job"), http.StatusBadRequest) return }
job, batch, err := s.db.GetJob(r.Context(), jobID) if errors.Is(err, db.ErrNotFound) { s.writeError(w, taskNotFound(), http.StatusNotFound) return } if err != nil { s.logger.Error("failed to read job", "err", err, "jobId", jobID) s.writeError(w, internalError, http.StatusInternalServerError) return } if batch.ID != input.TaskId || batch.OwnerDid != owner { s.writeError(w, taskNotFound(), http.StatusNotFound) return } if !job.Status.CanRetry() { s.writeError(w, jobNotRetryable(job.Status), http.StatusBadRequest) return }
var newEncryptedToken *string if job.Private { expired := batch.CredentialExpiresAt == nil || !batch.CredentialExpiresAt.After(time.Now()) switch { case input.GithubToken != nil && *input.GithubToken != "": aad := crypto.ComputeAAD(batch.OwnerDid, batch.ID, batch.RequestID) enc, err := crypto.Encrypt(s.cfg.ParsedMasterKey, *input.GithubToken, aad) if err != nil { s.logger.Error("failed to encrypt githubToken", "err", err) s.writeError(w, internalError, http.StatusInternalServerError) return } newEncryptedToken = &enc case batch.EncryptedToken == nil || *batch.EncryptedToken == "": s.writeError(w, invalidRequest("githubToken is required to retry a private job"), http.StatusBadRequest) return case expired: s.writeError(w, invalidRequest("the github credential for this task has expired; retry with a fresh one"), http.StatusBadRequest) return } }
parentBatch, batchJobs, err := s.db.RequeueJob(r.Context(), jobID, owner, newEncryptedToken) if errors.Is(err, db.ErrNotFound) { s.writeError(w, taskNotFound(), http.StatusNotFound) return } var transition *db.InvalidTransitionError if errors.As(err, &transition) { s.writeError(w, jobNotRetryable(transition.Status), http.StatusBadRequest) return } if err != nil { s.logger.Error("failed to retry job", "err", err, "jobId", jobID) s.writeError(w, internalError, http.StatusInternalServerError) return }
s.writeTask(w, parentBatch.Task(batchJobs), http.StatusOK)}
func mapSlice[T, U any](items []T, f func(T) U) []U { out := make([]U, len(items)) for i, it := range items { out[i] = f(it) } return out}
func taskView(task db.Task) *org_tangled.TempMigratorDefs_Task { updatedAt := task.UpdatedAt.UTC().Format(time.RFC3339)
return &org_tangled.TempMigratorDefs_Task{ Id: task.ID, OwnerDid: task.OwnerDid, CreatedAt: task.CreatedAt.UTC().Format(time.RFC3339), UpdatedAt: &updatedAt, Jobs: mapSlice(task.Jobs, jobView), }}
func jobView(job db.Job) *org_tangled.TempMigratorDefs_Job { var repoDid *string if job.RepoDid != "" { repoDid = &job.RepoDid }
return &org_tangled.TempMigratorDefs_Job{ Id: strconv.FormatInt(job.ID, 10), Name: job.Name, RepoDid: repoDid, KnotDid: job.KnotDid, SourceUrl: job.SourceURL, Private: job.Private, Status: string(job.Status), Attempts: int64(job.Attempts), Error: job.Error, }}
func DescriptionOf(job *org_tangled.TempMigratorDefs_NewJob) string { if job == nil || job.Description == nil { return "" } return strings.TrimSpace(*job.Description)}
func actorDid(r *http.Request) (string, bool) { did, ok := r.Context().Value(serviceauth.ActorDid).(syntax.DID) if !ok { return "", false } return did.String(), true}
func (s *Server) writeTask(w http.ResponseWriter, task db.Task, status int) { s.writeJSON(w, taskView(task), status)}
func (s *Server) writeJSON(w http.ResponseWriter, payload any, status int) { w.Header().Set("Content-Type", "application/json") w.Header().Set("Cache-Control", "no-store, no-cache, must-revalidate") w.WriteHeader(status) _ = json.NewEncoder(w).Encode(payload)}
func (s *Server) writeError(w http.ResponseWriter, e xrpcerr.XrpcError, status int) { s.writeJSON(w, e, status)}
func decodeStrictJSON(w http.ResponseWriter, r *http.Request, dst any, maxBytes int64) error { ct := r.Header.Get("Content-Type") if ct == "" || !strings.HasPrefix(strings.ToLower(ct), "application/json") { return errors.New("Content-Type must be application/json") }
if maxBytes <= 0 { maxBytes = 1 << 20 }
r.Body = http.MaxBytesReader(w, r.Body, maxBytes) dec := json.NewDecoder(r.Body) dec.DisallowUnknownFields()
if err := dec.Decode(dst); err != nil { return err }
if dec.More() { return errors.New("request body contains multiple JSON values") } var trailing struct{} if err := dec.Decode(&trailing); err != io.EOF { return errors.New("request body contains unexpected trailing characters") }
return nil}
func invalidRequest(message string) xrpcerr.XrpcError { return xrpcerr.NewXrpcError(xrpcerr.WithTag("InvalidRequest"), xrpcerr.WithMessage(message))}
func taskNotFound() xrpcerr.XrpcError { return xrpcerr.NewXrpcError( xrpcerr.WithTag("TaskNotFound"), xrpcerr.WithMessage("no such task belongs to you"), )}
func jobNotRetryable(status db.JobStatus) xrpcerr.XrpcError { return xrpcerr.NewXrpcError( xrpcerr.WithTag("JobNotRetryable"), xrpcerr.WithMessage(fmt.Sprintf("cannot retry a job with status %q", status)), )}
func taskConflict() xrpcerr.XrpcError { return xrpcerr.NewXrpcError( xrpcerr.WithTag("TaskConflict"), xrpcerr.WithMessage("requestId was used before with different jobs"), )}
func grantRequired(serviceURL, did string) xrpcerr.XrpcError { start := fmt.Sprintf("%s/oauth/start?did=%s&return_to=%s", strings.TrimSuffix(serviceURL, "/"), did, url.QueryEscape(config.DefaultReturnPath)) return xrpcerr.NewXrpcError( xrpcerr.WithTag("GrantRequired"), xrpcerr.WithMessage(fmt.Sprintf("grant the migrator access at %s", start)), )}
var ( missingActorError = xrpcerr.NewXrpcError( xrpcerr.WithTag("Auth"), xrpcerr.WithMessage("a verified service token is required"), ) internalError = xrpcerr.NewXrpcError( xrpcerr.WithTag("InternalError"), xrpcerr.WithMessage("internal error"), ))