internal/server/server.go
1
package server
3
import (
4
"context"
5
"errors"
6
"flag"
7
"fmt"
8
"io"
9
"net"
10
"net/http"
11
"net/netip"
12
"net/url"
13
"os"
14
"os/signal"
15
"strings"
16
"syscall"
17
"time"
19
sdk "github.com/modelcontextprotocol/go-sdk/mcp"
21
"github.com/koment-dev/koment/internal/anchor"
22
"github.com/koment-dev/koment/internal/application"
23
"github.com/koment-dev/koment/internal/auth"
24
"github.com/koment-dev/koment/internal/config"
25
githubprovider "github.com/koment-dev/koment/internal/github"
26
"github.com/koment-dev/koment/internal/listen"
27
"github.com/koment-dev/koment/internal/mcp"
28
"github.com/koment-dev/koment/internal/metrics"
29
"github.com/koment-dev/koment/internal/serving"
30
"github.com/koment-dev/koment/internal/store"
31
"github.com/koment-dev/koment/internal/ui"
32
)
34
const (
35
defaultAddress = "127.0.0.1:8080"
36
defaultSyncInterval = time.Minute
37
initialSyncTimeout = time.Minute
38
shutdownGrace = 10 * time.Second
39
headerTimeout = 10 * time.Second
40
requestTimeout = 2 * time.Minute
41
maximumMCPRequest = 1 << 20
42
maximumMutationBody = 1 << 20
43
minimumSyncInterval = 10 * time.Second
44
)
46
const usage = `koment serve presents authenticated human and agent views of Git repositories.
48
koment serve --config <repositories.yaml> [--listen <addr>]
50
The UI, repository switcher and MCP endpoint read the same immutable commit
51
snapshots. Liveness and readiness are public; all source and rationale routes
52
require a trusted proxy identity, a scoped bearer credential, or a loopback
53
listener. Remote writes are available only when a GitHub token is configured.
54
`
56
func Serve(args []string, stderr io.Writer) error {
57
flags := flag.NewFlagSet("serve", flag.ContinueOnError)
58
flags.SetOutput(stderr)
59
flags.Usage = func() {
60
fmt.Fprint(stderr, usage, "\nFlags (each also settable from the environment):\n", config.Usage(flags))
61
}
62
configurationPath := flags.String("config", "", "strict YAML repository configuration")
63
address := flags.String("listen", defaultAddress, "authenticated UI and MCP listen address")
64
metricsAddress := flags.String("metrics", "", "separate unauthenticated metrics address; off unless given")
65
githubTokenFile := flags.String("github-token-file", "", "file containing the GitHub App or fine-grained token")
66
credentialsFile := flags.String("credentials-file", "", "secret YAML file containing hashed scoped agent credentials")
67
trustedProxies := flags.String("trusted-proxies", "", "comma-separated CIDR ranges allowed to assert human identity headers")
68
humanWrites := flags.Bool("human-writes", false, "allow trusted-proxy humans to create reviewed annotations")
69
syncInterval := flags.Duration("sync-interval", defaultSyncInterval, "repository refresh interval")
70
if err := flags.Parse(args); err != nil {
71
return err
72
}
73
if err := config.FromEnvironment(flags); err != nil {
74
return err
75
}
76
if flags.NArg() != 0 {
77
return fmt.Errorf("serve takes no arguments, got %s", flags.Arg(0))
78
}
79
if *configurationPath == "" {
80
return errors.New("serve needs --config")
81
}
82
if *syncInterval < minimumSyncInterval {
83
return fmt.Errorf("sync interval %s is below the %s minimum", *syncInterval, minimumSyncInterval)
84
}
86
repositories, err := loadRepositories(*configurationPath)
87
if err != nil {
88
return err
89
}
90
catalog, err := serving.NewCatalog(repositories)
91
if err != nil {
92
return err
93
}
94
resolvedAddress, err := listen.Address(*address)
95
if err != nil {
96
return err
97
}
98
proxyRanges, err := parsePrefixes(*trustedProxies)
99
if err != nil {
100
return err
101
}
102
credentials := auth.Credentials{}
103
if *credentialsFile != "" {
104
credentials, err = auth.LoadCredentials(*credentialsFile)
105
if err != nil {
106
return err
107
}
108
}
109
allowLoopback := listen.IsLoopback(resolvedAddress)
110
if !allowLoopback && len(proxyRanges) == 0 && len(credentials.Tokens) == 0 {
111
return errors.New("a non-loopback server needs --trusted-proxies or --credentials-file")
112
}
113
authenticator, err := auth.New(auth.Configuration{
114
AllowLoopback: allowLoopback, TrustedProxies: proxyRanges,
115
HumanCanWrite: *humanWrites, CredentialStore: credentials,
116
})
117
if err != nil {
118
return err
119
}
120
githubToken := ""
121
if *githubTokenFile != "" {
122
content, readErr := readBounded(*githubTokenFile, maximumConfiguration)
123
if readErr != nil {
124
return fmt.Errorf("reading GitHub token: %w", readErr)
125
}
126
githubToken = strings.TrimSpace(string(content))
127
if githubToken == "" {
128
return errors.New("GitHub token file is empty")
129
}
130
}
131
provider := githubprovider.New(githubToken)
132
synchronizer := serving.Synchronizer{Catalog: catalog, Source: provider}
133
initialContext, cancelInitial := context.WithTimeout(context.Background(), initialSyncTimeout)
134
if syncErr := synchronizer.RefreshAll(initialContext); syncErr != nil {
135
fmt.Fprintf(stderr, "koment: initial synchronization incomplete: %v\n", syncErr)
136
}
137
cancelInitial()
139
mainListener, err := net.Listen("tcp", resolvedAddress)
140
if err != nil {
141
return fmt.Errorf("listening on %s: %w", resolvedAddress, err)
142
}
143
var metricsListener net.Listener
144
if *metricsAddress != "" {
145
metricsResolved, resolveErr := listen.Address(*metricsAddress)
146
if resolveErr != nil {
147
return errors.Join(resolveErr, mainListener.Close())
148
}
149
metricsListener, err = net.Listen("tcp", metricsResolved)
150
if err != nil {
151
return errors.Join(fmt.Errorf("listening for metrics on %s: %w", metricsResolved, err), mainListener.Close())
152
}
153
}
155
rootContext, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
156
defer stop()
157
ctx, cancel := context.WithCancel(rootContext)
158
defer cancel()
159
recorder := metrics.Recorder(metrics.Discard{})
160
var metricSet *metrics.Metrics
161
if metricsListener != nil {
162
metricSet = metrics.New()
163
recorder = metricSet
164
}
165
go synchronize(ctx, synchronizer, *syncInterval, recorder, stderr)
166
var materializer serving.Materializer
167
if githubToken != "" {
168
materializer = provider
169
}
170
application := &http.Server{
171
Handler: metrics.Instrument(recorder, "serve", Handler(catalog, authenticator, recorder, materializer)),
172
ReadHeaderTimeout: headerTimeout, ReadTimeout: requestTimeout, WriteTimeout: requestTimeout,
173
}
174
var metricsServer *http.Server
175
if metricsListener != nil {
176
metricsServer = &http.Server{Handler: metricSet.Handler(), ReadHeaderTimeout: headerTimeout}
177
}
178
fmt.Fprintf(stderr, "koment: serving UI and MCP at http://%s\n", mainListener.Addr())
179
if metricsListener != nil {
180
fmt.Fprintf(stderr, "koment: metrics at http://%s/metrics\n", metricsListener.Addr())
181
}
182
return serveListeners(ctx, cancel, application, mainListener, metricsServer, metricsListener, stderr)
183
}
185
func Handler(
186
catalog *serving.Catalog, authenticator *auth.Authenticator, recorder metrics.Recorder, materializer serving.Materializer,
187
) http.Handler {
188
accessFromRequest := func(request *http.Request) map[string]bool {
189
principal, found := auth.FromContext(request.Context())
190
if !found {
191
return map[string]bool{}
192
}
193
return repositoryAccess(catalog, principal, auth.Read)
194
}
195
canWrite := func(request *http.Request, repository string) bool {
196
principal, found := auth.FromContext(request.Context())
197
return found && materializer != nil && principal.Can(repository, auth.Write)
198
}
199
human := ui.SnapshotHandlerCapabilities(catalog, accessFromRequest, canWrite)
200
agent := sdk.NewStreamableHTTPHandler(func(request *http.Request) *sdk.Server {
201
principal, _ := auth.FromContext(request.Context())
202
readAccess := repositoryAccess(catalog, principal, auth.Read)
203
if materializer != nil && principal.Permissions[auth.Write] {
204
return mcp.NewWritableSnapshotServer(catalog, recorder, mcp.RepositoryAccess(readAccess), principal.Author(), materializer)
205
}
206
return mcp.NewSnapshotServer(catalog, recorder, mcp.RepositoryAccess(readAccess))
207
}, &sdk.StreamableHTTPOptions{
208
Stateless: true, JSONResponse: true, MaxRequestBodyBytes: maximumMCPRequest,
209
PropagateRequestCancellation: true,
210
})
211
protected := http.NewServeMux()
212
protected.Handle("/mcp", agent)
213
protected.HandleFunc("POST /r/{repository}/annotations", func(writer http.ResponseWriter, request *http.Request) {
214
addRemoteAnnotation(writer, request, catalog, materializer)
215
})
216
protected.Handle("/", human)
217
routes := http.NewServeMux()
218
routes.HandleFunc("GET /livez", func(writer http.ResponseWriter, _ *http.Request) {
219
writer.Header().Set("Content-Type", "text/plain; charset=utf-8")
220
fmt.Fprintln(writer, "ok")
221
})
222
routes.HandleFunc("GET /readyz", func(writer http.ResponseWriter, _ *http.Request) {
223
if err := catalog.Ready(); err != nil {
224
http.Error(writer, err.Error(), http.StatusServiceUnavailable)
225
return
226
}
227
writer.Header().Set("Content-Type", "text/plain; charset=utf-8")
228
fmt.Fprintln(writer, "ready")
229
})
230
crossOrigin := http.NewCrossOriginProtection()
231
routes.Handle("/", authenticator.Middleware(crossOrigin.Handler(protected)))
232
return securityHeaders(routes)
233
}
235
func addRemoteAnnotation(
236
writer http.ResponseWriter, request *http.Request, catalog *serving.Catalog, materializer serving.Materializer,
237
) {
238
principal, authenticated := auth.FromContext(request.Context())
239
repositoryID := request.PathValue("repository")
240
if !authenticated || materializer == nil || !principal.Can(repositoryID, auth.Write) {
241
http.Error(writer, "repository write access denied", http.StatusForbidden)
242
return
243
}
244
state, found := catalog.State(repositoryID)
245
if !found {
246
http.Error(writer, "repository not found", http.StatusNotFound)
247
return
248
}
249
if state.Snapshot == nil {
250
http.Error(writer, "repository has no synchronized snapshot", http.StatusServiceUnavailable)
251
return
252
}
253
request.Body = http.MaxBytesReader(writer, request.Body, maximumMutationBody)
254
if err := request.ParseForm(); err != nil {
255
http.Error(writer, "invalid annotation form: "+err.Error(), http.StatusBadRequest)
256
return
257
}
258
kind, err := store.ParseType(request.Form.Get("kind"))
259
if err != nil {
260
http.Error(writer, err.Error(), http.StatusBadRequest)
261
return
262
}
263
record, err := application.DraftAnnotation(state.Snapshot, application.AddInput{
264
File: request.Form.Get("file"), Excerpt: request.Form.Get("excerpt"),
265
Kind: kind, Body: request.Form.Get("body"), Author: principal.Author(),
266
})
267
if err != nil {
268
http.Error(writer, err.Error(), http.StatusBadRequest)
269
return
270
}
271
review, err := materializer.Materialize(request.Context(), state.Repository, state.Snapshot.Commit, record)
272
if err != nil {
273
http.Error(writer, err.Error(), http.StatusBadGateway)
274
return
275
}
276
query := url.Values{"created": []string{record.Metadata.ID}, "review": []string{review.URL}}
277
target := "/r/" + url.PathEscape(repositoryID) + "/f/" + escapeSourcePath(record.Spec.Target.File) + "?" + query.Encode()
278
http.Redirect(writer, request, target, http.StatusSeeOther)
279
}
281
func escapeSourcePath(sourcePath string) string {
282
parts := strings.Split(sourcePath, "/")
283
for index, part := range parts {
284
parts[index] = url.PathEscape(part)
285
}
286
return strings.Join(parts, "/")
287
}
289
func repositoryAccess(catalog *serving.Catalog, principal auth.Principal, permission auth.Permission) map[string]bool {
290
access := make(map[string]bool)
291
for _, repository := range catalog.Repositories() {
292
if principal.Can(repository.Identity.ID, permission) {
293
access[repository.Identity.ID] = true
294
}
295
}
296
return access
297
}
299
func securityHeaders(next http.Handler) http.Handler {
300
return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
301
writer.Header().Set("Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'")
302
writer.Header().Set("Referrer-Policy", "no-referrer")
303
writer.Header().Set("X-Content-Type-Options", "nosniff")
304
writer.Header().Set("X-Frame-Options", "DENY")
305
next.ServeHTTP(writer, request)
306
})
307
}
309
func synchronize(ctx context.Context, synchronizer serving.Synchronizer, interval time.Duration, recorder metrics.Recorder, stderr io.Writer) {
310
ticker := time.NewTicker(interval)
311
defer ticker.Stop()
312
observeCatalog(synchronizer.Catalog, recorder, 0)
313
var refreshableAt time.Time
314
for {
315
select {
316
case <-ctx.Done():
317
return
318
case <-ticker.C:
319
if now := time.Now().UTC(); now.Before(refreshableAt) {
320
continue
321
}
322
started := time.Now()
323
if err := synchronizer.RefreshAll(ctx); err != nil {
324
fmt.Fprintf(stderr, "koment: synchronization: %v\n", err)
325
if until, limited := githubprovider.RetryAfter(err); limited {
326
refreshableAt = until
327
fmt.Fprintf(stderr, "koment: synchronization paused until %s\n", until.Format(time.RFC3339))
328
}
329
} else {
330
refreshableAt = time.Time{}
331
}
332
observeCatalog(synchronizer.Catalog, recorder, time.Since(started))
333
}
334
}
335
}
337
func observeCatalog(catalog *serving.Catalog, recorder metrics.Recorder, duration time.Duration) {
338
counts := make(map[anchor.Status]int)
339
files := 0
340
for _, state := range catalog.States() {
341
if state.Snapshot == nil {
342
continue
343
}
344
files += len(state.Snapshot.Files)
345
for status, count := range state.Snapshot.Counts() {
346
counts[status] += count
347
}
348
}
349
recorder.ObserveRepository(counts, files, duration)
350
}
352
func serveListeners(
353
ctx context.Context, cancel context.CancelFunc, application *http.Server, applicationListener net.Listener,
354
metricsServer *http.Server, metricsListener net.Listener, stderr io.Writer,
355
) error {
356
type result struct {
357
name string
358
err error
359
}
360
listeners := 1
361
results := make(chan result, 2)
362
go func() { results <- result{name: "application", err: application.Serve(applicationListener)} }()
363
if metricsServer != nil {
364
listeners++
365
go func() { results <- result{name: "metrics", err: metricsServer.Serve(metricsListener)} }()
366
}
367
go func() {
368
<-ctx.Done()
369
shutdownContext, stop := context.WithTimeout(context.Background(), shutdownGrace)
370
defer stop()
371
if err := application.Shutdown(shutdownContext); err != nil {
372
fmt.Fprintf(stderr, "koment: shutting down application: %v\n", err)
373
}
374
if metricsServer != nil {
375
if err := metricsServer.Shutdown(shutdownContext); err != nil {
376
fmt.Fprintf(stderr, "koment: shutting down metrics: %v\n", err)
377
}
378
}
379
}()
380
first := <-results
381
cancel()
382
if first.err != nil && !errors.Is(first.err, http.ErrServerClosed) {
383
return fmt.Errorf("%s listener: %w", first.name, first.err)
384
}
385
for range listeners - 1 {
386
remaining := <-results
387
if remaining.err != nil && !errors.Is(remaining.err, http.ErrServerClosed) {
388
return fmt.Errorf("%s listener: %w", remaining.name, remaining.err)
389
}
390
}
391
return nil
392
}
394
func parsePrefixes(specification string) ([]netip.Prefix, error) {
395
if strings.TrimSpace(specification) == "" {
396
return nil, nil
397
}
398
var prefixes []netip.Prefix
399
for _, entry := range strings.Split(specification, ",") {
400
prefix, err := netip.ParsePrefix(strings.TrimSpace(entry))
401
if err != nil {
402
return nil, fmt.Errorf("trusted proxy %q is not a CIDR range: %w", entry, err)
403
}
404
prefixes = append(prefixes, prefix.Masked())
405
}
406
return prefixes, nil
407
}