coverage — ~bigbes/sr-ht-dolt — 3523280c — remoteapi/interceptors.go

Coverage
63.5% 73/115 statements
Δ
+0.0
Blob
0d29122
Uncovered L120-L122L124-L138L146-L151L183-L185L189-L192L195-L196L265-L269L276-L279L314-L317L320-L322L329-L331L359-L360L363-L364L373-L379L381-L391L406L410-L417
1 package remoteapi
2
3 import (
4 "context"
5 "database/sql"
6 "errors"
7 "fmt"
8 "log/slog"
9
10 remotesapi "github.com/dolthub/dolt/go/gen/proto/dolt/services/remotesapi/v1alpha1"
11 "github.com/vaughan0/go-ini"
12 "google.golang.org/grpc"
13 "google.golang.org/grpc/codes"
14 "google.golang.org/grpc/metadata"
15 "google.golang.org/grpc/status"
16
17 "go.bigb.es/auxilia/scribe"
18
19 "sourcecraft.dev/bigbes/sr-ht-core/auth"
20 "sourcecraft.dev/bigbes/sr-ht-core/config"
21 "sourcecraft.dev/bigbes/sr-ht-core/database"
22
23 "sourcecraft.dev/bigbes/sr-ht-dolt/authn"
24 "sourcecraft.dev/bigbes/sr-ht-dolt/core"
25 "sourcecraft.dev/bigbes/sr-ht-dolt/db"
26 "sourcecraft.dev/bigbes/sr-ht-dolt/storage"
27 )
28
29 // Method-classification sets, copied verbatim from upstream remotesrv's
30 // interceptors.go (SUPER_USER_RPC_METHODS / CLONE_ADMIN_RPC_METHODS). Upstream
31 // drops the authenticated context and never sees the repo path, so it can only
32 // make a binary superuser/clone-admin decision; we keep its exact method lists
33 // but layer our own per-repo ACL check on top (see authorize).
34 //
35 // - writeMethods are pushes: they require OpPush / AccessRW.
36 // - readMethods are clone/pull/fetch reads: OpCloneRead / AccessRO.
37 // - anything else is an unknown method and is denied (PermissionDenied),
38 // matching upstream's "unknown rpc method" hard failure.
39 var (
40 writeMethods = map[string]bool{
41 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/AddTableFiles": true,
42 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/Commit": true,
43 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/GetUploadLocations": true,
44 }
45 readMethods = map[string]bool{
46 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/GetDownloadLocations": true,
47 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/GetRepoMetadata": true,
48 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/HasChunks": true,
49 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/ListTableFiles": true,
50 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/RefreshTableFileUrl": true,
51 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/Root": true,
52 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/StreamDownloadLocations": true,
53 "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/StreamChunkLocations": true,
54 }
55
56 // rootMethod is the only read RPC the dolt client may send with no repo path
57 // (a bare "what is the current root" ping used during dial/handshake). It is
58 // treated as an unauthenticated-OK ping: it still authenticates the caller
59 // (so a bad token is rejected) but skips the per-repo ACL check when no path
60 // is present. Every real read carries a repo path and is checked normally.
61 rootMethod = "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/Root"
62 )
63
64 // repoRequest is the subset of every ChunkStoreService request message that
65 // carries the target repository, mirrored from upstream remotesrv's private
66 // repoRequest interface. All request types implement it.
67 type repoRequest interface {
68 GetRepoId() *remotesapi.RepoId
69 GetRepoPath() string
70 }
71
72 // repoStore is the narrow slice of db.Store the interceptor needs. Declaring it
73 // locally (rather than depending on *db.Store directly) keeps the authorization
74 // logic unit-testable with an in-memory stub — no Postgres required.
75 type repoStore interface {
76 GetRepoByOwnerAndName(ctx context.Context, ownerUsername, name string) (*core.Repo, error)
77 EffectiveAccess(ctx context.Context, userID, repoID int) (*core.AccessMode, error)
78 // CreateRepo and DeleteRepo back push-to-create: an authenticated caller
79 // touching a not-yet-existing repo in their own namespace has the row
80 // inserted (CreateRepo) and, if the on-disk store then fails to materialize,
81 // rolled back (DeleteRepo).
82 CreateRepo(ctx context.Context, r *core.Repo) (*core.Repo, error)
83 DeleteRepo(ctx context.Context, id int) error
84 }
85
86 // interceptor holds the collaborators the per-RPC auth/authz decision needs. It
87 // is installed on the remotesrv gRPC server via Options().
88 type interceptor struct {
89 conf ini.File
90 service string
91 expectedAud string
92 keys authn.KeyStore
93 logger *slog.Logger
94
95 // pool is the shared database pool, threaded into the request context via
96 // database.Context so db.FromContext-style lookups and the token resolvers
97 // can reach it.
98 pool *sql.DB
99
100 // stores returns a repoStore for a request. Production binds a fresh
101 // db.Store to the shared pool; tests inject a stub.
102 stores func() repoStore
103
104 // reposRoot is the absolute directory under which bare NBS stores live; it
105 // resolves the on-disk path for a push-to-created repository.
106 reposRoot string
107
108 // createStore materializes the empty on-disk store for a push-to-created
109 // repository. Production uses storage.InitEmptyStore; tests inject a fake.
110 createStore func(ctx context.Context, absPath string) error
111 }
112
113 // newInterceptor builds the interceptor over the shared pool. It binds a fresh
114 // db.Store per request (the pool owns connection lifetime, ctx bounds each
115 // query). pool and keys must be non-nil.
116 //
117 // The logger is not a parameter: it is slog's default with this component's
118 // name on it. Threading one through the constructor was logrus' requirement,
119 // not this package's — there is no configuration here a caller ever varied.
120 0 func newInterceptor(conf ini.File, service, expectedAud string, pool *sql.DB, keys authn.KeyStore, reposRoot string, createStore func(ctx context.Context, absPath string) error) *interceptor {
121 0 if pool == nil {
122 0 panic("remoteapi: newInterceptor requires a non-nil *sql.DB")
123 }
124 0 if createStore == nil {
125 0 createStore = storage.InitEmptyStore
126 0 }
127 0 i := &interceptor{
128 0 conf: conf,
129 0 service: service,
130 0 expectedAud: expectedAud,
131 0 keys: keys,
132 0 logger: slog.Default().With("component", "remotesapi.auth"),
133 0 pool: pool,
134 0 reposRoot: reposRoot,
135 0 createStore: createStore,
136 0 }
137 0 i.stores = func() repoStore { return db.NewStore(pool) }
138 0 return i
139 }
140
141 // dbHandle returns the shared pool threaded into request contexts.
142 1 func (i *interceptor) dbHandle() *sql.DB { return i.pool }
143
144 // Options returns the gRPC server options that install both interceptors,
145 // matching upstream ServerInterceptor.Options.
146 0 func (i *interceptor) Options() []grpc.ServerOption {
147 0 return []grpc.ServerOption{
148 0 grpc.ChainUnaryInterceptor(i.unary()),
149 0 grpc.ChainStreamInterceptor(i.stream()),
150 0 }
151 0 }
152
153 // classify maps a full gRPC method to its access op/mode. ok is false for an
154 // unknown method, which the caller denies.
155 34 func classify(fullMethod string) (op core.Op, mode core.AccessMode, ok bool) {
156 34 if writeMethods[fullMethod] {
157 16 return core.OpPush, core.AccessRW, true
158 16 }
159 18 if readMethods[fullMethod] {
160 16 return core.OpCloneRead, core.AccessRO, true
161 16 }
162 2 return 0, "", false
163 }
164
165 // withServiceCtx augments the live per-RPC context (which carries the gRPC
166 // deadline, cancellation and incoming metadata) with the config and database
167 // values the authn resolvers and db.Store read from context. We augment the
168 // handler's context rather than starting from a stored base context so request
169 // deadlines and the incoming "authorization" metadata are preserved.
170 1 func (i *interceptor) withServiceCtx(ctx context.Context) context.Context {
171 1 ctx = config.Context(ctx, i.conf, i.service)
172 1 ctx = database.Context(ctx, i.dbHandle())
173 1 return ctx
174 1 }
175
176 // authenticate resolves the caller from the incoming "authorization" metadata.
177 // It returns the caller (nil for an anonymous request) or a gRPC status error:
178 // Unauthenticated for a bad/forged/revoked credential, Unavailable for a
179 // transient backend failure (meta.sr.ht or the database unreachable).
180 1 func (i *interceptor) authenticate(ctx context.Context) (*auth.AuthContext, error) {
181 1 header := ""
182 1 if md, ok := metadata.FromIncomingContext(ctx); ok {
183 0 if vals := md.Get("authorization"); len(vals) > 0 {
184 0 header = vals[0]
185 0 }
186 }
187 1 ac, err := authn.ResolveGRPCAuth(ctx, header, i.expectedAud, i.keys)
188 1 if err != nil {
189 0 if errors.Is(err, authn.ErrInvalidToken) {
190 0 i.logger.WarnContext(ctx, "authentication rejected", scribe.Err(err))
191 0 return nil, status.Error(codes.Unauthenticated, "invalid or expired credentials")
192 0 }
193 // Transient: meta.sr.ht or the database is unreachable. Never surface as a
194 // hard credential rejection — the client should retry.
195 0 i.logger.ErrorContext(ctx, "authentication backend error", scribe.Err(err))
196 0 return nil, status.Error(codes.Unavailable, "authentication temporarily unavailable")
197 }
198 1 return ac, nil
199 }
200
201 // authorize applies the grant gate and the per-repo ACL to an authenticated
202 // caller and returns a context carrying the resolved caller for the handler, or
203 // a gRPC status error. It is called once per unary request and once per stream
204 // message. The order mirrors the plan:
205 //
206 // 1. classify the method (unknown ⇒ PermissionDenied);
207 // 2. OAuth grant gate (TokenGrantsAllow) — PAT callers must carry
208 // dolt.sr.ht/repos:RO for reads / :RW for pushes; anonymous, cookie and
209 // dolt-key callers pass trivially;
210 // 3. extract the repo path (Root with no path ⇒ unauthenticated-OK ping);
211 // 4. load the repository row and the caller's effective ACL;
212 // 5. core.Allowed — on denial, NotFound for a PRIVATE repo the caller cannot
213 // even see (no existence leak), PermissionDenied for a visible repo.
214 28 func (i *interceptor) authorize(ctx context.Context, ac *auth.AuthContext, fullMethod string, req any) (context.Context, error) {
215 28 op, mode, ok := classify(fullMethod)
216 28 if !ok {
217 1 return nil, status.Errorf(codes.PermissionDenied, "unknown rpc method: %s", fullMethod)
218 1 }
219
220 // Grant gate is a global property of the token (independent of any repo), so
221 // running it first leaks nothing about repository existence.
222 27 if !authn.TokenGrantsAllow(ac, mode) {
223 2 return nil, status.Errorf(codes.PermissionDenied,
224 2 "token grants do not permit %s on %s repositories", mode, i.service)
225 2 }
226
227 25 rr, isRepoReq := req.(repoRequest)
228 25 repoPath := ""
229 25 if isRepoReq {
230 25 repoPath = repoPathOf(rr)
231 25 }
232 25 if repoPath == "" {
233 2 if fullMethod == rootMethod {
234 1 // Handshake ping with no repo: authenticated but not repo-scoped.
235 1 return authn.WithCaller(ctx, ac), nil
236 1 }
237 1 return nil, status.Error(codes.InvalidArgument, "request is missing a repository path")
238 }
239
240 23 owner, name, err := core.ParseRepoPath(repoPath)
241 23 if err != nil {
242 1 return nil, status.Errorf(codes.InvalidArgument, "invalid repository path %q: %v", repoPath, err)
243 1 }
244
245 22 caller := authn.AsCoreCaller(ac)
246 22
247 22 store := i.stores()
248 22 repo, err := store.GetRepoByOwnerAndName(ctx, owner, name)
249 22 if err != nil {
250 9 if errors.Is(err, db.ErrNotFound) {
251 9 // Push-to-create: an authenticated, non-suspended caller touching a
252 9 // not-yet-existing repo in THEIR OWN namespace (with a valid name)
253 9 // has it transparently created — a PRIVATE row plus a genuinely
254 9 // empty on-disk store — then proceeds through the normal ACL check
255 9 // as the owner. Any other case (anonymous, suspended, another
256 9 // user's namespace, invalid name) keeps returning NotFound, which
257 9 // also avoids leaking existence of a stranger's private repos.
258 9 if caller == nil || caller.Suspended || owner != caller.Username || core.ValidateName(name) != nil {
259 5 return nil, status.Errorf(codes.NotFound, "repository %s/%s not found", owner, name)
260 5 }
261 4 repo, err = i.autoCreate(ctx, store, caller, owner, name)
262 4 if err != nil {
263 1 return nil, err
264 1 }
265 0 } else {
266 0 i.logger.ErrorContext(ctx, "repository lookup failed",
267 0 "owner", owner, "name", name, scribe.Err(err))
268 0 return nil, status.Error(codes.Unavailable, "repository lookup temporarily unavailable")
269 0 }
270 }
271
272 16 var aclMode *core.AccessMode
273 16 if caller != nil {
274 11 aclMode, err = store.EffectiveAccess(ctx, caller.UserID, repo.ID)
275 11 if err != nil {
276 0 i.logger.ErrorContext(ctx, "effective-access lookup failed",
277 0 "user_id", caller.UserID, "repo_id", repo.ID, scribe.Err(err))
278 0 return nil, status.Error(codes.Unavailable, "authorization temporarily unavailable")
279 0 }
280 }
281
282 16 if !core.Allowed(caller, repo, aclMode, op) {
283 4 if core.NotFoundForPrivate(caller, repo, aclMode) {
284 2 return nil, status.Errorf(codes.NotFound, "repository %s/%s not found", owner, name)
285 2 }
286 2 return nil, status.Errorf(codes.PermissionDenied, "%s denied on %s/%s", op, owner, name)
287 }
288
289 12 return authn.WithCaller(ctx, ac), nil
290 }
291
292 // autoCreate transparently creates a PRIVATE repository owned by caller under
293 // owner/name (push-to-create). It inserts the row and materializes a genuinely
294 // empty on-disk store. A concurrent first-push that already created the row is
295 // tolerated: the caller loses the CreateRepo race, re-fetches the winner's row,
296 // and does NOT re-create the store. If the store fails to materialize, the row
297 // is rolled back so a later push retries cleanly. Preconditions (authenticated,
298 // non-suspended, owns the namespace, valid name) are checked by the caller.
299 4 func (i *interceptor) autoCreate(ctx context.Context, store repoStore, caller *core.Caller, owner, name string) (*core.Repo, error) {
300 4 newRepo := &core.Repo{
301 4 Name: name,
302 4 OwnerID: caller.UserID,
303 4 OwnerName: owner,
304 4 Path: storage.RepoDiskPath(i.reposRoot, owner, name),
305 4 Visibility: core.VisibilityPrivate,
306 4 }
307 4 created, cerr := store.CreateRepo(ctx, newRepo)
308 4 if cerr != nil {
309 1 if errors.Is(cerr, db.ErrNameTaken) {
310 1 // Lost the race: another first-push created the row (and its store).
311 1 // Adopt the winner's row; do not touch disk.
312 1 repo, gerr := store.GetRepoByOwnerAndName(ctx, owner, name)
313 1 if gerr != nil {
314 0 i.logger.ErrorContext(ctx, "auto-create refetch failed",
315 0 "owner", owner, "name", name, scribe.Err(gerr))
316 0 return nil, status.Error(codes.Unavailable, "repository creation temporarily unavailable")
317 0 }
318 1 return repo, nil
319 }
320 0 i.logger.ErrorContext(ctx, "auto-create failed",
321 0 "owner", owner, "name", name, scribe.Err(cerr))
322 0 return nil, status.Error(codes.Unavailable, "repository creation temporarily unavailable")
323 }
324
325 3 if serr := i.createStore(ctx, created.Path); serr != nil {
326 1 // The row exists but the store does not: roll the row back (best effort)
327 1 // so the repo does not linger half-created and a retry can succeed.
328 1 if derr := store.DeleteRepo(ctx, created.ID); derr != nil {
329 0 i.logger.ErrorContext(ctx, "auto-create rollback failed",
330 0 "owner", owner, "name", name, "repo_id", created.ID, scribe.Err(derr))
331 0 }
332 1 i.logger.ErrorContext(ctx, "auto-create store initialization failed",
333 1 "owner", owner, "name", name, scribe.Err(serr))
334 1 return nil, status.Error(codes.Unavailable, "repository creation temporarily unavailable")
335 }
336 2 return created, nil
337 }
338
339 // repoPathOf extracts the target repo path from a request without panicking on
340 // an absent path (upstream's getRepoPath panics on empty path + nil repo id).
341 // It returns "" when neither is present so the caller can treat Root specially.
342 25 func repoPathOf(req repoRequest) string {
343 25 if p := req.GetRepoPath(); p != "" {
344 22 return p
345 22 }
346 3 if id := req.GetRepoId(); id != nil {
347 1 return fmt.Sprintf("%s/%s", id.Org, id.RepoName)
348 1 }
349 2 return ""
350 }
351
352 // unary is the unary server interceptor: authenticate once, authorize the
353 // request, then invoke the handler with the caller-carrying context.
354 1 func (i *interceptor) unary() grpc.UnaryServerInterceptor {
355 1 return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
356 1 ctx = i.withServiceCtx(ctx)
357 1 ac, err := i.authenticate(ctx)
358 1 if err != nil {
359 0 return nil, err
360 0 }
361 1 authedCtx, err := i.authorize(ctx, ac, info.FullMethod, req)
362 1 if err != nil {
363 0 return nil, err
364 0 }
365 1 return handler(authedCtx, req)
366 }
367 }
368
369 // stream is the stream server interceptor. Authentication happens once at
370 // stream start; the repo-path authorization is re-checked for every message the
371 // client sends (the streaming reads carry a repo path per message), matching the
372 // plan's "override Context() and check each RecvMsg" design.
373 0 func (i *interceptor) stream() grpc.StreamServerInterceptor {
374 0 return func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
375 0 ctx := i.withServiceCtx(ss.Context())
376 0 ac, err := i.authenticate(ctx)
377 0 if err != nil {
378 0 return err
379 0 }
380 // A stream to an unknown method is denied before the handler runs.
381 0 if _, _, ok := classify(info.FullMethod); !ok {
382 0 return status.Errorf(codes.PermissionDenied, "unknown rpc method: %s", info.FullMethod)
383 0 }
384 0 wrapped := &authStream{
385 0 ServerStream: ss,
386 0 ctx: authn.WithCaller(ctx, ac),
387 0 i: i,
388 0 ac: ac,
389 0 fullMethod: info.FullMethod,
390 0 }
391 0 return handler(srv, wrapped)
392 }
393 }
394
395 // authStream wraps a grpc.ServerStream so the handler sees the caller-carrying
396 // context and every received message is re-authorized against its repo path.
397 type authStream struct {
398 grpc.ServerStream
399 ctx context.Context
400 i *interceptor
401 ac *auth.AuthContext
402 fullMethod string
403 }
404
405 // Context returns the authenticated, caller-carrying context.
406 0 func (s *authStream) Context() context.Context { return s.ctx }
407
408 // RecvMsg receives the next message and re-authorizes it against its repo path
409 // before handing it to the handler. An authorization failure aborts the stream.
410 0 func (s *authStream) RecvMsg(m any) error {
411 0 if err := s.ServerStream.RecvMsg(m); err != nil {
412 0 return err
413 0 }
414 0 if _, err := s.i.authorize(s.ctx, s.ac, s.fullMethod, m); err != nil {
415 0 return err
416 0 }
417 0 return nil
418 }