Compare commits

..
Author SHA1 Message Date
mathiasandClaude Opus 5.5 87fa28e2c8 fix(update,session): brain_update and session_log confine caller paths (#169)
CI / Lint / Test / Vet (pull_request) Successful in 45s
CI / Mirror to GitHub (pull_request) Skipped
Review round 1 on PR #170 found two more handlers with the #169 class of
bug (verified by execution):

- brain_update resolved its caller path with resolveWithin, a lexical
  containment check: any brain file could be superseded
  (classification.yaml, raw/, sessions/), and a final symlink was
  written through. UpdateNote now resolves like brain_edit
  (notes.Resolve, ScopeNotes: knowledge/ and wiki/ .md notes), and reads
  and rewrites through brainfs (ReadRegular, WriteRegular).
- session_log joined the caller's session_id onto the sessions dir and
  opened it O_CREATE|O_APPEND: an id with path segments created or
  appended a .jsonl file anywhere the pod could write. The id must now
  be a plain name (^[A-Za-z0-9_][A-Za-z0-9._-]{0,127}$, which every live
  session file matches), and the open adds O_NOFOLLOW|O_NONBLOCK. Read
  checks the id too.
- Tunnel tests: the .txt case now names an existing file (it failed on
  "missing" before), the symlink test adds an in-brain target, and the
  FIFO comment no longer claims the handle check is the only guard.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 22:38:08 +02:00
mathiasandClaude Opus 5.5 9eb3d2456d fix(tunnel): brain_tunnel cannot write outside the wiki or through symlinks
CI / Lint / Test / Vet (pull_request) Successful in 52s
CI / Mirror to GitHub (pull_request) Skipped
brain_tunnel passes caller-supplied source and target paths to
WriteTunnel, which checked only the first segments ("wiki/<wing>/...")
and joined them onto the brain dir. "wiki/b/x/../../../../tmp/f.md"
resolved outside the brain, and appendSeeAlso read and rewrote that file
(verified by execution): any MCP token holder could append to any
existing file the pod user can write, inside the brain (raw/, sessions/,
classification.yaml) or outside it.

- checkNotePath: only a clean, local wiki/.../<slug>.md path, with no
  ".", ".." or empty segments.
- Both notes are read and rewritten through brainfs. New
  brainfs.WriteRegular: never creates; Lstat through the root, open with
  O_NOFOLLOW|O_NONBLOCK, and require the opened handle to be the same
  regular file (os.SameFile). os.Root.OpenFile follows an in-root final
  symlink even with O_NOFOLLOW (verified on go1.27), so the flag alone
  does not refuse it.
- The source is no longer rewritten when the target is refused.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 22:23:45 +02:00
33 changed files with 484 additions and 1916 deletions
-26
View File
@@ -1,26 +0,0 @@
package main
import (
"context"
"fmt"
"log/slog"
chassisauth "git.d-ma.be/mathias/mcp-chassis/auth"
)
// buildJWTValidator builds the JWT validator for issuer (nil when issuer is
// empty: JWT auth disabled). Since mcp-chassis v0.7 construction no longer
// fails when OIDC discovery does (mcp-chassis#4): the issuer resolves lazily,
// and until it does a JWT is answered 503-and-retry, not 401. That replaces
// #83's boot-time retry loop (hyperguild#155). An issuer still unresolved at
// boot is logged, so a misconfigured issuer is visible rather than silent.
func buildJWTValidator(ctx context.Context, logger *slog.Logger, issuer, audience string) (*chassisauth.JWTValidator, error) {
v, err := chassisauth.NewJWTValidator(ctx, issuer, audience)
if err != nil {
return nil, fmt.Errorf("build jwt validator: %w", err)
}
if ok, rerr := v.Ready(ctx); !ok {
logger.Warn("jwt issuer not resolved at boot; JWTs answer 503 until it is", "issuer", issuer, "err", rerr)
}
return v, nil
}
-111
View File
@@ -1,111 +0,0 @@
package main
import (
"bytes"
"context"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/metrics"
)
// fakeIssuer answers OIDC discovery with 502 for the first fail requests, as
// ingress does while dex is not yet serving on a cold boot (#83), then with a
// real discovery document and an empty JWKS.
func fakeIssuer(t *testing.T, fail int32) (*httptest.Server, *atomic.Int32) {
t.Helper()
var discoveries atomic.Int32
mux := http.NewServeMux()
var srv *httptest.Server
mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, _ *http.Request) {
if discoveries.Add(1) <= fail {
w.WriteHeader(http.StatusBadGateway)
return
}
w.Header().Set("Content-Type", "application/json")
_, _ = io.WriteString(w, `{"issuer":"`+srv.URL+`","jwks_uri":"`+srv.URL+`/keys"}`)
})
mux.HandleFunc("/keys", func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = io.WriteString(w, `{"keys":[]}`)
})
srv = httptest.NewServer(mux)
t.Cleanup(srv.Close)
return srv, &discoveries
}
// #83 retried discovery at boot because chassis v0.2.0 failed construction on
// a 502. Since chassis v0.7 (mcp-chassis#4) the validator is built anyway and
// resolves the issuer lazily, so boot no longer waits or exits on it
// (hyperguild#155). An unresolved issuer is logged at boot.
func TestBuildJWTValidatorReturnsAValidatorWhileDiscovery502s(t *testing.T) {
srv, _ := fakeIssuer(t, 1000)
var buf bytes.Buffer
v, err := buildJWTValidator(context.Background(), slog.New(slog.NewTextHandler(&buf, nil)), srv.URL, "brain")
require.NoError(t, err)
assert.NotNil(t, v)
assert.Contains(t, buf.String(), "jwt issuer not resolved at boot")
}
func TestBuildJWTValidatorLogsNothingAgainstAHealthyIssuer(t *testing.T) {
srv, _ := fakeIssuer(t, 0)
var buf bytes.Buffer
v, err := buildJWTValidator(context.Background(), slog.New(slog.NewTextHandler(&buf, nil)), srv.URL, "brain")
require.NoError(t, err)
assert.NotNil(t, v)
assert.NotContains(t, buf.String(), "not resolved")
}
func TestBuildJWTValidatorIsNilWithoutAnIssuer(t *testing.T) {
v, err := buildJWTValidator(context.Background(), slog.New(slog.DiscardHandler), "", "brain")
require.NoError(t, err)
assert.Nil(t, v)
}
// With the issuer unresolved, a JWT on a REST route answers 503 (retry), not
// 401: the token may be fine, the server just cannot check it yet.
func TestBuildMuxAnswers503ToAJWTWhileTheIssuerIsUnresolved(t *testing.T) {
srv, _ := fakeIssuer(t, 1000)
v, err := buildJWTValidator(context.Background(), slog.New(slog.DiscardHandler), srv.URL, "brain")
require.NoError(t, err)
mux, err := buildMux(restAuth{token: routesTok, validator: v, realm: "brain"}, testHandler(t),
okHandler("mcp"), metrics.New(), nil)
require.NoError(t, err)
code, _ := do(t, mux, "GET", "/pending", "Bearer aaa.bbb.ccc")
assert.Equal(t, http.StatusServiceUnavailable, code)
}
// PR #183 review: /mcp shares REST's bearer, but only the static token was
// asserted on it. With the issuer unresolved, a JWT on /mcp answers 503 too,
// so /mcp's middleware was built with the JWT validator.
func TestBuildMuxAnswers503ToAJWTOnMCPWhileTheIssuerIsUnresolved(t *testing.T) {
srv, _ := fakeIssuer(t, 1000)
v, err := buildJWTValidator(context.Background(), slog.New(slog.DiscardHandler), srv.URL, "brain")
require.NoError(t, err)
mux, err := buildMux(restAuth{token: routesTok, validator: v, realm: "brain"}, testHandler(t),
okHandler("mcp"), metrics.New(), nil)
require.NoError(t, err)
code, _ := do(t, mux, "POST", "/mcp", "Bearer aaa.bbb.ccc")
assert.Equal(t, http.StatusServiceUnavailable, code)
}
// PR #183 review: a 401 on /mcp names the protected-resource metadata in
// WWW-Authenticate (what the claude.ai OAuth connector discovers from), so
// /mcp's middleware was built with resourceMetadataURL.
func TestBuildMuxMCP401CarriesResourceMetadataInWWWAuthenticate(t *testing.T) {
const rm = "https://brain.example/.well-known/oauth-protected-resource"
mux, err := buildMux(restAuth{token: routesTok, realm: "brain", resourceMetadataURL: rm}, testHandler(t),
okHandler("mcp"), metrics.New(), nil)
require.NoError(t, err)
rec := httptest.NewRecorder()
mux.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/mcp", nil))
require.Equal(t, http.StatusUnauthorized, rec.Code)
assert.Contains(t, rec.Header().Get("WWW-Authenticate"), `resource_metadata="`+rm+`"`)
}
+39 -23
View File
@@ -32,6 +32,7 @@ import (
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline" "github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
"github.com/mathiasbq/hyperguild/ingestion/internal/reranker" "github.com/mathiasbq/hyperguild/ingestion/internal/reranker"
"github.com/mathiasbq/hyperguild/ingestion/internal/search" "github.com/mathiasbq/hyperguild/ingestion/internal/search"
"github.com/mathiasbq/hyperguild/ingestion/internal/startupretry"
"github.com/mathiasbq/hyperguild/ingestion/internal/symlinkguard" "github.com/mathiasbq/hyperguild/ingestion/internal/symlinkguard"
"github.com/mathiasbq/hyperguild/ingestion/internal/vectorstore" "github.com/mathiasbq/hyperguild/ingestion/internal/vectorstore"
"github.com/mathiasbq/hyperguild/ingestion/internal/watcher" "github.com/mathiasbq/hyperguild/ingestion/internal/watcher"
@@ -368,15 +369,27 @@ func main() {
// hyperguild#157: report a symlink that appears in the brain while running. // hyperguild#157: report a symlink that appears in the brain while running.
go symlinkguard.Watch(ctx, brainDir, symlinkScanInterval(os.Getenv), buildNotifier()) go symlinkguard.Watch(ctx, brainDir, symlinkScanInterval(os.Getenv), buildNotifier())
// Never fatal on an unreachable issuer (#83): on a cold cluster boot this // Built before the watcher, the graph backfill and session ingestion, and
// pod can beat dex to serving and get a 502 from ingress, and exiting meant // retried rather than fatal (#83). Two separate reasons:
// CrashLoopBackOff redoing the backfill. Since mcp-chassis v0.7 the //
// validator resolves the issuer lazily and answers JWTs 503-and-retry until // - On a cold cluster boot this pod can beat dex to serving and gets a
// it does (hyperguild#155), so no boot retry is needed. A misconfigured // 502 from ingress. Nothing is misconfigured; it simply started first.
// issuer no longer exits the process: it is logged at boot and every JWT // Exiting there meant CrashLoopBackOff redoing the backfill on every
// gets 503; the static bearer keeps working. // attempt, with exponential backoff stretching the brain outage.
jwtValidator, err := buildJWTValidator(ctx, logger, // - Constructing it first means a genuine config error (wrong issuer,
os.Getenv("DEX_ISSUER_URL"), os.Getenv("MCP_AUDIENCE")) // bad audience) fails in seconds instead of after all that work.
//
// The budget stays bounded: retrying forever would turn a misconfiguration
// into a process that looks alive and never serves.
jwtValidator, err := startupretry.Do(ctx,
startupretry.Policy{Attempts: 8, Base: time.Second, Max: 30 * time.Second},
func(ctx context.Context) (*chassisauth.JWTValidator, error) {
v, err := chassisauth.NewJWTValidator(ctx, os.Getenv("DEX_ISSUER_URL"), os.Getenv("MCP_AUDIENCE"))
if err != nil {
logger.Warn("oidc discovery not ready, retrying", "err", err)
}
return v, err
})
if err != nil { if err != nil {
logger.Error("build jwt validator", "err", err) logger.Error("build jwt validator", "err", err)
os.Exit(1) os.Exit(1)
@@ -516,6 +529,8 @@ func main() {
} }
} }
public["/mcp"] = chassisauth.BearerMiddleware(mcpToken, jwtValidator, "brain", resourceMetadataURL, mcpSrv)
// POST /capture (#53/#54): the uniform capture REST door. Needs a ticket // POST /capture (#53/#54): the uniform capture REST door. Needs a ticket
// tracker to file action items, so it only mounts when Gitea is // tracker to file action items, so it only mounts when Gitea is
// configured. It reuses the MCP server's graph-wired brain store (one // configured. It reuses the MCP server's graph-wired brain store (one
@@ -539,7 +554,7 @@ func main() {
public["POST /capture"] = captureH public["POST /capture"] = captureH
// Same use-case behind the MCP `capture` tool (#55 relay) so MCP-native // Same use-case behind the MCP `capture` tool (#55 relay) so MCP-native
// harnesses (claude.ai, Crush, Pi, LLM Council) reach capture through // harnesses (claude.ai, Crush, Pi, LLM Council) reach capture through
// the existing /mcp OAuth connector. buildMux wraps this same mcpSrv; // the existing /mcp OAuth connector. mcpSrv is already wrapped above;
// WithCapture mutates the same instance, so the tool appears live. // WithCapture mutates the same instance, so the tool appears live.
mcpSrv.WithCapture(captureSvc, jwtValidator, mcpToken, "local-cli", resolver) mcpSrv.WithCapture(captureSvc, jwtValidator, mcpToken, "local-cli", resolver)
// hyperguild#97: a governance-file rewrite gets the same trail a // hyperguild#97: a governance-file rewrite gets the same trail a
@@ -581,18 +596,19 @@ func main() {
os.Exit(1) os.Exit(1)
} }
// /metrics — unauthenticated Prometheus endpoint, scraped via the // /metrics — unauthenticated Prometheus endpoint. kube-prometheus-stack
// ServiceMonitor in k3s/apps/supervisor/. buildMux registers it and wraps // scrapes it via the ServiceMonitor in k3s/apps/supervisor/. The metrics
// the whole router in the registry's middleware. // middleware below wraps every other registered handler so it observes
// // real request latency. /metrics itself is excluded from its own
// The whole router, /mcp's bearer included, is built in buildMux so its // observation by registering it on the outer mux (post-wrap).
// tests are tests of production wiring (PR #152 review, hyperguild#154). reg := metrics.New()
// The REST API's credential is the same as /mcp's. Untested here: the public["GET /metrics"] = http.HandlerFunc(reg.Handler())
// restAuth fields and the public map this call passes in (e.g. a nil
// validator would drop JWT auth on /mcp and REST and the suite stays // The whole router, built in one place so the test of buildMux is a test
// green), and ListenAndServe serving the result unchanged (hyperguild#182). // of production wiring (PR #152 review). The REST API's credential is the
handler, err := buildMux(restAuth{token: mcpToken, validator: jwtValidator, realm: "brain", // same as /mcp's.
resourceMetadataURL: resourceMetadataURL}, h, mcpSrv, metrics.New(), public) mux, err := buildMux(restAuth{token: mcpToken, validator: jwtValidator, realm: "brain",
resourceMetadataURL: resourceMetadataURL}, h, public)
if err != nil { if err != nil {
logger.Error("build router", "err", err) logger.Error("build router", "err", err)
os.Exit(1) os.Exit(1)
@@ -613,7 +629,7 @@ func main() {
"watch_interval", watchIntervalLog, "watch_interval", watchIntervalLog,
"mcp_enabled", true, "mcp_enabled", true,
) )
if err := http.ListenAndServe(addr, handler); err != nil { if err := http.ListenAndServe(addr, reg.Middleware(mux)); err != nil {
logger.Error("server stopped", "err", err) logger.Error("server stopped", "err", err)
os.Exit(1) os.Exit(1)
} }
+19 -51
View File
@@ -8,22 +8,22 @@ import (
chassisauth "git.d-ma.be/mathias/mcp-chassis/auth" chassisauth "git.d-ma.be/mathias/mcp-chassis/auth"
"github.com/mathiasbq/hyperguild/ingestion/internal/api" "github.com/mathiasbq/hyperguild/ingestion/internal/api"
"github.com/mathiasbq/hyperguild/ingestion/internal/metrics"
) )
// publicRoutes is the complete set of patterns a caller may register on the // publicRoutes is the complete set of patterns allowed on the outer mux,
// outer mux, outside the REST API's auth (hyperguild#149). Each authenticates // outside the REST API's auth (hyperguild#149). Each authenticates itself or is
// itself or is meant to be open: /capture (bearer/JWT), the webhook (HMAC), // meant to be open: /mcp (bearer/JWT), /capture (bearer/JWT), the webhook
// /oauth/token (client secret) and OAuth metadata (open). /mcp and /metrics // (HMAC), /oauth/token (client secret), OAuth metadata and /metrics (open).
// are not here: buildMux builds them itself (hyperguild#154). A REST pattern // A REST pattern added here would beat the authenticated catch-all by
// added here would beat the authenticated catch-all by specificity and ship // specificity and ship open, which is why buildMux refuses anything else.
// open, which is why buildMux refuses anything else.
var publicRoutes = map[string]bool{ var publicRoutes = map[string]bool{
"/mcp": true,
"POST /capture": true, "POST /capture": true,
"POST /webhooks/brain-sync": true, "POST /webhooks/brain-sync": true,
"GET /.well-known/oauth-protected-resource": true, "GET /.well-known/oauth-protected-resource": true,
"GET /.well-known/oauth-authorization-server": true, "GET /.well-known/oauth-authorization-server": true,
"POST /oauth/token": true, "POST /oauth/token": true,
"GET /metrics": true,
} }
// restAuth is the credential the REST API checks: the same as /mcp's. // restAuth is the credential the REST API checks: the same as /mcp's.
@@ -34,25 +34,16 @@ type restAuth struct {
resourceMetadataURL string resourceMetadataURL string
} }
// buildMux builds the handler main passes to ListenAndServe unchanged. The // buildMux is the server's whole router. The REST API is mounted at "/"
// REST API is mounted at "/" and /mcp is registered, both behind bearer auth // behind bearer auth that buildMux constructs itself, so a caller cannot pass
// that buildMux constructs itself from auth (one credential), so a caller // a no-op; public handlers are registered as given, and only for patterns on
// cannot pass a no-op. GET /metrics serves reg, and the whole router is // the publicRoutes allowlist. It returns an http.Handler, not the
// wrapped in reg's middleware. Other public handlers are registered as given, // *ServeMux, so nothing can register a route on the result afterwards and
// and only for patterns on the publicRoutes allowlist. It returns an // bypass the allowlist.
// http.Handler, not the *ServeMux, so nothing can register a route on the func buildMux(auth restAuth, h *api.Handler, public map[string]http.Handler) (http.Handler, error) {
// result. What stays untested is main's call site: the restAuth fields and
// the public map main passes in, and ListenAndServe serving this handler
// unchanged (hyperguild#182). The tests build their own arguments.
func buildMux(
auth restAuth, h *api.Handler, mcpHandler http.Handler, reg *metrics.Registry, public map[string]http.Handler,
) (http.Handler, error) {
if auth.token == "" { if auth.token == "" {
return nil, errors.New("buildMux: REST auth needs a token") return nil, errors.New("buildMux: REST auth needs a token")
} }
if mcpHandler == nil || reg == nil {
return nil, errors.New("buildMux: needs the MCP handler and the metrics registry")
}
mux := http.NewServeMux() mux := http.NewServeMux()
for pattern, handler := range public { for pattern, handler := range public {
if !publicRoutes[pattern] { if !publicRoutes[pattern] {
@@ -60,15 +51,10 @@ func buildMux(
} }
mux.Handle(pattern, handler) mux.Handle(pattern, handler)
} }
bearer := func(next http.Handler) http.Handler { mountREST(mux, h, func(next http.Handler) http.Handler {
return chassisauth.BearerMiddleware(auth.token, auth.validator, auth.realm, auth.resourceMetadataURL, next) return chassisauth.BearerMiddleware(auth.token, auth.validator, auth.realm, auth.resourceMetadataURL, next)
} })
mux.Handle("/mcp", bearer(mcpHandler)) return mux, nil
// /metrics is registered on the mux and so observed by its own
// middleware, like any route.
mux.Handle("GET /metrics", reg.Handler())
mountREST(mux, h, bearer)
return reg.Middleware(mux), nil
} }
// mountREST registers the REST API on its own mux and mounts it at "/" behind // mountREST registers the REST API on its own mux and mounts it at "/" behind
@@ -87,23 +73,5 @@ func mountREST(mux *http.ServeMux, h *api.Handler, auth func(http.Handler) http.
rest.HandleFunc("POST /promote", h.Promote) rest.HandleFunc("POST /promote", h.Promote)
rest.HandleFunc("POST /backfill-embeddings", h.BackfillEmbeddings) rest.HandleFunc("POST /backfill-embeddings", h.BackfillEmbeddings)
rest.HandleFunc("GET /pass-rate", h.PassRate) rest.HandleFunc("GET /pass-rate", h.PassRate)
mux.Handle("/", labelRESTPattern(rest, auth(rest))) mux.Handle("/", auth(rest))
}
// labelRESTPattern sets the request's Pattern to the REST route it addresses
// before auth runs, so the metrics middleware (which reads Pattern from the
// same *http.Request) labels a refused request by its route rather than by
// the "/" mount (hyperguild#155). A refused request to a path no route
// matches keeps "/", so for refused requests the label set stays bounded by
// the registered routes. That bound does not hold for authenticated
// requests: an unregistered path reaches the REST mux, which 404s with an
// empty Pattern, and the metrics middleware falls back to the raw URL path
// (hyperguild#185).
func labelRESTPattern(rest *http.ServeMux, next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if _, pattern := rest.Handler(r); pattern != "" {
r.Pattern = pattern
}
next.ServeHTTP(w, r)
})
} }
@@ -1,88 +0,0 @@
package main
import (
"bytes"
"log/slog"
"net/http"
"net/http/httptest"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// wrongTokSentinel is a presented credential distinctive enough that any
// occurrence in a log line is the raw token, not a coincidence.
const wrongTokSentinel = "wrong-tok-sentinel-q7zx"
// hyperguild#155: since #149 every REST route sits behind BearerMiddleware,
// and chassis v0.2.0 refused without logging, so an unknown caller got
// silent 401s. The chassis now writes an `mcp auth rejected` line, and the
// presented token appears nowhere in the log as a bare value (PR #181
// review: asserting on the "Bearer " header form could never fail, since
// the chassis logs fields, not headers).
func TestBuildMuxRejectionLogLineOmitsTheBarePresentedToken(t *testing.T) {
var buf bytes.Buffer
prev := slog.Default()
slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil)))
t.Cleanup(func() { slog.SetDefault(prev) })
code, _ := do(t, testMux(t), "POST", "/query", "Bearer "+wrongTokSentinel)
require.Equal(t, http.StatusUnauthorized, code)
assert.Contains(t, buf.String(), "mcp auth rejected")
assert.Contains(t, buf.String(), "path=/query")
assert.NotContains(t, buf.String(), wrongTokSentinel, "the rejected token is logged as a bare value")
assert.NotContains(t, buf.String(), routesTok, "the valid static token is logged")
}
// The `mcp auth accepted` line leaves out the valid static token.
func TestBuildMuxAcceptedLogLineOmitsTheBareStaticToken(t *testing.T) {
var buf bytes.Buffer
prev := slog.Default()
slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil)))
t.Cleanup(func() { slog.SetDefault(prev) })
code, _ := do(t, testMux(t), "GET", "/pending", "Bearer "+routesTok)
require.Equal(t, http.StatusOK, code)
assert.Contains(t, buf.String(), "mcp auth accepted")
assert.NotContains(t, buf.String(), routesTok, "the accepted static token is logged as a bare value")
}
// An authenticated REST request is labelled with the REST mux's pattern:
// the inner mux sets it on the same *http.Request the outer middleware reads.
func TestMetricsLabelAnAuthenticatedRESTRequestWithItsRoutePattern(t *testing.T) {
mux, reg := testMuxAndRegistry(t)
code, _ := do(t, mux, "GET", "/pending", "Bearer "+routesTok)
require.Equal(t, http.StatusOK, code)
rec := httptest.NewRecorder()
reg.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
assert.Contains(t, rec.Body.String(), `path="GET /pending"`)
assert.NotContains(t, rec.Body.String(), `path="/"`)
}
// hyperguild#155: a request refused by REST auth never reaches the REST mux,
// so it was labelled path="/" and a 4xx on /query could not be told from one
// on /write. It is now labelled with the REST pattern it was addressed to.
func TestMetricsLabelARejectedRESTRequestWithItsRoutePattern(t *testing.T) {
mux, reg := testMuxAndRegistry(t)
code, _ := do(t, mux, "POST", "/query", "")
require.Equal(t, http.StatusUnauthorized, code)
rec := httptest.NewRecorder()
reg.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
assert.Contains(t, rec.Body.String(), `path="POST /query",status="4xx"`)
}
// A path no REST route matches keeps the catch-all label, so an attacker
// probing random URLs cannot grow the series set.
func TestMetricsLabelARejectedUnknownPathWithTheCatchAll(t *testing.T) {
mux, reg := testMuxAndRegistry(t)
code, _ := do(t, mux, "POST", "/no-such-route-zq", "")
require.Equal(t, http.StatusUnauthorized, code)
rec := httptest.NewRecorder()
reg.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
assert.Contains(t, rec.Body.String(), `path="/",status="4xx"`)
assert.NotContains(t, rec.Body.String(), "no-such-route-zq")
}
+12 -67
View File
@@ -12,7 +12,6 @@ import (
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/api" "github.com/mathiasbq/hyperguild/ingestion/internal/api"
"github.com/mathiasbq/hyperguild/ingestion/internal/metrics"
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline" "github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
) )
@@ -22,34 +21,20 @@ func okHandler(body string) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(body)) }) return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(body)) })
} }
// testMux builds the router through buildMux, the function main calls, with // testMux builds the router exactly as main does: buildMux with the full
// every allowlisted public pattern (stub handlers), a stub MCP server and a // public set main registers (stub handlers) and real bearer config.
// static token only: no JWT validator and no resourceMetadataURL (the tests // hyperguild#149: the REST API is authenticated as a whole; the PR #152
// in jwt_validator_test.go pass those). It does not use main's own restAuth // review showed a test of a COPY of the router let main.go regress unseen.
// or public map (hyperguild#182). hyperguild#149: the REST API is
// authenticated as a whole; the PR #152 review showed a test of a COPY of
// the router let main.go regress unseen.
func testMux(t *testing.T) http.Handler { func testMux(t *testing.T) http.Handler {
t.Helper() t.Helper()
mux, _ := testMuxAndRegistry(t) h := api.NewHandler(t.TempDir(), slog.New(slog.NewTextHandler(os.Stderr, nil)), pipeline.Config{})
return mux
}
func testMuxAndRegistry(t *testing.T) (http.Handler, *metrics.Registry) {
t.Helper()
public := map[string]http.Handler{} public := map[string]http.Handler{}
for p := range publicRoutes { for p := range publicRoutes {
public[p] = okHandler("public " + p) public[p] = okHandler("public " + p)
} }
reg := metrics.New() mux, err := buildMux(restAuth{token: routesTok, realm: "brain"}, h, public)
mux, err := buildMux(restAuth{token: routesTok, realm: "brain"}, testHandler(t), okHandler("mcp"), reg, public)
require.NoError(t, err) require.NoError(t, err)
return mux, reg return mux
}
func testHandler(t *testing.T) *api.Handler {
t.Helper()
return api.NewHandler(t.TempDir(), slog.New(slog.NewTextHandler(os.Stderr, nil)), pipeline.Config{})
} }
func do(t *testing.T, h http.Handler, method, path, authz string) (int, string) { func do(t *testing.T, h http.Handler, method, path, authz string) (int, string) {
@@ -89,9 +74,8 @@ func TestBuildMuxAuthenticatesAnUnregisteredPath(t *testing.T) {
} }
// buildMux registers public handlers as given, without adding auth: each // buildMux registers public handlers as given, without adding auth: each
// public route owns its own (bearer/JWT on /capture, HMAC on the webhook, // public route owns its own (bearer on /mcp, HMAC on the webhook, client
// client secret on /oauth/token) or is meant to be open (OAuth metadata). // secret on /oauth/token) or is meant to be open (/metrics, OAuth metadata).
// /mcp and /metrics are not public routes: buildMux builds them itself.
func TestBuildMuxDoesNotWrapPublicRoutes(t *testing.T) { func TestBuildMuxDoesNotWrapPublicRoutes(t *testing.T) {
r := testMux(t) r := testMux(t)
for p := range publicRoutes { for p := range publicRoutes {
@@ -109,53 +93,14 @@ func TestBuildMuxDoesNotWrapPublicRoutes(t *testing.T) {
// over the authenticated catch-all (more specific pattern), so it is refused. // over the authenticated catch-all (more specific pattern), so it is refused.
func TestBuildMuxRefusesAPublicPatternOffTheAllowlist(t *testing.T) { func TestBuildMuxRefusesAPublicPatternOffTheAllowlist(t *testing.T) {
h := api.NewHandler(t.TempDir(), slog.New(slog.NewTextHandler(os.Stderr, nil)), pipeline.Config{}) h := api.NewHandler(t.TempDir(), slog.New(slog.NewTextHandler(os.Stderr, nil)), pipeline.Config{})
// "/mcp" and "GET /metrics" are built by buildMux itself (hyperguild#154): for _, p := range []string{"POST /query", "GET /pending", "/", "POST /metrics"} {
// passing either as public would register them unauthenticated or twice. _, err := buildMux(restAuth{token: routesTok, realm: "brain"}, h, map[string]http.Handler{p: okHandler("x")})
for _, p := range []string{"POST /query", "GET /pending", "/", "POST /metrics", "/mcp", "GET /metrics"} {
_, err := buildMux(restAuth{token: routesTok, realm: "brain"}, h, okHandler("mcp"), metrics.New(),
map[string]http.Handler{p: okHandler("x")})
assert.Error(t, err, "public pattern %q must be refused", p) assert.Error(t, err, "public pattern %q must be refused", p)
} }
} }
func TestBuildMuxRefusesAnEmptyToken(t *testing.T) { func TestBuildMuxRefusesAnEmptyToken(t *testing.T) {
h := api.NewHandler(t.TempDir(), slog.New(slog.NewTextHandler(os.Stderr, nil)), pipeline.Config{}) h := api.NewHandler(t.TempDir(), slog.New(slog.NewTextHandler(os.Stderr, nil)), pipeline.Config{})
_, err := buildMux(restAuth{token: "", realm: "brain"}, h, okHandler("mcp"), metrics.New(), nil) _, err := buildMux(restAuth{token: "", realm: "brain"}, h, nil)
assert.Error(t, err) assert.Error(t, err)
} }
// hyperguild#154: /mcp's bearer was built in main.go, untested; the buildMux
// tests used stubs. buildMux now wraps the MCP handler itself, with the same
// credential as the REST API.
func TestBuildMuxRefusesMCPWithoutAValidToken(t *testing.T) {
r := testMux(t)
for _, authz := range []string{"", "Bearer wrong"} {
code, body := do(t, r, "POST", "/mcp", authz)
assert.Equal(t, http.StatusUnauthorized, code, "authz %q", authz)
assert.NotEqual(t, "mcp", body)
}
code, body := do(t, r, "POST", "/mcp", "Bearer "+routesTok)
assert.Equal(t, http.StatusOK, code)
assert.Equal(t, "mcp", body)
}
func TestBuildMuxRefusesANilMCPHandlerOrRegistry(t *testing.T) {
auth := restAuth{token: routesTok, realm: "brain"}
_, err := buildMux(auth, testHandler(t), nil, metrics.New(), nil)
assert.Error(t, err)
_, err = buildMux(auth, testHandler(t), okHandler("mcp"), nil, nil)
assert.Error(t, err)
}
// hyperguild#154: main wrapped buildMux's result in the metrics middleware
// before ListenAndServe, untested. buildMux now returns the handler main
// serves as-is, metrics included: a request through it is observed and
// exposed on its own GET /metrics.
func TestBuildMuxObservesRequestsAndServesMetrics(t *testing.T) {
r := testMux(t)
code, _ := do(t, r, "GET", "/pending", "Bearer "+routesTok)
require.Equal(t, http.StatusOK, code)
code, body := do(t, r, "GET", "/metrics", "")
require.Equal(t, http.StatusOK, code)
assert.Contains(t, body, `path="GET /pending"`)
}
+7 -7
View File
@@ -34,9 +34,9 @@ require (
github.com/rogpeppe/go-internal v1.15.0 // indirect github.com/rogpeppe/go-internal v1.15.0 // indirect
github.com/spf13/pflag v1.0.5 // indirect github.com/spf13/pflag v1.0.5 // indirect
github.com/x448/float16 v0.8.4 // indirect github.com/x448/float16 v0.8.4 // indirect
golang.org/x/net v0.55.0 // indirect golang.org/x/net v0.26.0 // indirect
golang.org/x/oauth2 v0.21.0 // indirect golang.org/x/oauth2 v0.21.0 // indirect
golang.org/x/term v0.43.0 // indirect golang.org/x/term v0.28.0 // indirect
golang.org/x/time v0.3.0 // indirect golang.org/x/time v0.3.0 // indirect
google.golang.org/protobuf v1.34.2 // indirect google.golang.org/protobuf v1.34.2 // indirect
gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
@@ -51,7 +51,7 @@ require (
) )
require ( require (
git.d-ma.be/mathias/mcp-chassis v0.7.1 git.d-ma.be/mathias/mcp-chassis v0.2.0
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect
github.com/goccy/go-json v0.10.3 // indirect github.com/goccy/go-json v0.10.3 // indirect
@@ -66,9 +66,9 @@ require (
github.com/lestrrat-go/option v1.0.1 // indirect github.com/lestrrat-go/option v1.0.1 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
github.com/segmentio/asm v1.2.0 // indirect github.com/segmentio/asm v1.2.0 // indirect
golang.org/x/crypto v0.51.0 // indirect golang.org/x/crypto v0.32.0 // indirect
golang.org/x/sync v0.21.0 // indirect golang.org/x/sync v0.17.0 // indirect
golang.org/x/sys v0.45.0 // indirect golang.org/x/sys v0.31.0 // indirect
golang.org/x/text v0.39.0 // indirect golang.org/x/text v0.29.0 // indirect
gopkg.in/yaml.v3 v3.0.1 gopkg.in/yaml.v3 v3.0.1
) )
+16 -16
View File
@@ -1,5 +1,5 @@
git.d-ma.be/mathias/mcp-chassis v0.7.1 h1:4h8TOBHTA0L/no4IPcr4tLcQtmKUlhE2/cXyPdNFGmM= git.d-ma.be/mathias/mcp-chassis v0.2.0 h1:6fLmb7xqRa2nNVWsHaUbbfbArgDXJw/gDhb09clBIjo=
git.d-ma.be/mathias/mcp-chassis v0.7.1/go.mod h1:Ks7EK2UnGAN0H3rJjKUxUagX8/ZBdtLrOlcUbv0RwH8= git.d-ma.be/mathias/mcp-chassis v0.2.0/go.mod h1:Ks7EK2UnGAN0H3rJjKUxUagX8/ZBdtLrOlcUbv0RwH8=
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -117,42 +117,42 @@ github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9dec
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI= golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8= golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= golang.org/x/net v0.26.0 h1:soB7SVo0PWrY4vPW/+ay0jKDNScG2X9wFeYlXIvJsOQ=
golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/net v0.26.0/go.mod h1:5YKkiSynbBIh3p6iOc/vibscux0x38BZDkn8sCUPxHE=
golang.org/x/oauth2 v0.21.0 h1:tsimM75w1tF/uws5rbeHzIWxEqElMehnc+iW793zsZs= golang.org/x/oauth2 v0.21.0 h1:tsimM75w1tF/uws5rbeHzIWxEqElMehnc+iW793zsZs=
golang.org/x/oauth2 v0.21.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI= golang.org/x/oauth2 v0.21.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4= golang.org/x/term v0.28.0 h1:/Ts8HFuMR2E6IP/jlo7QVLZHggjKQbhu/7H0LJFr3Gg=
golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= golang.org/x/term v0.28.0/go.mod h1:Sw/lC2IAUZ92udQNf3WodGtn4k/XoLyZoh8v/8uiwek=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus= golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM= golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4= golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q= golang.org/x/tools v0.36.0 h1:kWS0uv/zsvHEle1LbV5LE8QujrxB3wfQyxHfhOk0Qkg=
golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA= golang.org/x/tools v0.36.0/go.mod h1:WBDiHKJK8YgLHlcQPYQzNCkUxUypCaa5ZegCVutKm+s=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+37 -67
View File
@@ -2,18 +2,12 @@ package brain
import ( import (
"bufio" "bufio"
"bytes"
"errors"
"fmt" "fmt"
"io"
"io/fs"
"os" "os"
"path/filepath" "path/filepath"
"sort" "sort"
"strings" "strings"
"time" "time"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
) )
// noteEntry is one row in a Wing _index.md. // noteEntry is one row in a Wing _index.md.
@@ -33,18 +27,13 @@ func BuildWingIndex(brainDir, wing string) error {
return fmt.Errorf("invalid wing %q", wing) return fmt.Errorf("invalid wing %q", wing)
} }
wingDir := filepath.Join(brainDir, "wiki", w) wingDir := filepath.Join(brainDir, "wiki", w)
// Listed and read through the brain root (hyperguild#157): a wing or if _, err := os.Stat(wingDir); os.IsNotExist(err) {
// hall that is a symlink is not followed, and notes are regular files.
bfs, err := brainfs.Open(brainDir)
if err != nil {
return fmt.Errorf("open brain: %w", err)
}
defer func() { _ = bfs.Close() }()
if _, err := bfs.Lstat("wiki/" + w); errors.Is(err, fs.ErrNotExist) {
return nil return nil
} else if err != nil {
return fmt.Errorf("stat wing: %w", err)
} }
entries, err := collectWingEntries(bfs, "wiki/"+w) entries, err := collectWingEntries(wingDir)
if err != nil { if err != nil {
return err return err
} }
@@ -69,47 +58,51 @@ func BuildWingIndex(brainDir, wing string) error {
// BuildAllWingIndexes regenerates _index.md for every wing under brain/wiki/. // BuildAllWingIndexes regenerates _index.md for every wing under brain/wiki/.
func BuildAllWingIndexes(brainDir string) error { func BuildAllWingIndexes(brainDir string) error {
bfs, err := brainfs.Open(brainDir) wikiDir := filepath.Join(brainDir, "wiki")
if err != nil { ents, err := os.ReadDir(wikiDir)
return fmt.Errorf("open brain: %w", err) if os.IsNotExist(err) {
}
wings, err := bfs.ReadDirDirs("wiki")
_ = bfs.Close()
if errors.Is(err, fs.ErrNotExist) {
return nil return nil
} }
if err != nil { if err != nil {
return fmt.Errorf("read wiki: %w", err) return fmt.Errorf("read wiki: %w", err)
} }
for _, w := range wings { for _, e := range ents {
if err := BuildWingIndex(brainDir, w); err != nil { if !e.IsDir() {
return fmt.Errorf("index %s: %w", w, err) continue
}
if err := BuildWingIndex(brainDir, e.Name()); err != nil {
return fmt.Errorf("index %s: %w", e.Name(), err)
} }
} }
return nil return nil
} }
func collectWingEntries(bfs *brainfs.Brain, wingRel string) ([]noteEntry, error) { func collectWingEntries(wingDir string) ([]noteEntry, error) {
var out []noteEntry var out []noteEntry
halls, err := bfs.ReadDirDirs(wingRel) ents, err := os.ReadDir(wingDir)
if err != nil { if err != nil {
return nil, fmt.Errorf("read wing: %w", err) return nil, fmt.Errorf("read wing: %w", err)
} }
for _, hall := range halls { for _, hallEnt := range ents {
if !hallEnt.IsDir() {
continue
}
hall := hallEnt.Name()
if !IsValidHall(hall) { if !IsValidHall(hall) {
continue continue
} }
hallRel := wingRel + "/" + hall hallDir := filepath.Join(wingDir, hall)
names, err := bfs.ReadDirRegular(hallRel) notes, err := os.ReadDir(hallDir)
if err != nil { if err != nil {
return nil, fmt.Errorf("read hall %s: %w", hall, err) return nil, fmt.Errorf("read hall %s: %w", hall, err)
} }
for _, name := range names { for _, n := range notes {
if !strings.HasSuffix(name, ".md") || name == "_index.md" { if n.IsDir() || !strings.HasSuffix(n.Name(), ".md") || n.Name() == "_index.md" {
continue continue
} }
slug := strings.TrimSuffix(name, ".md") slug := strings.TrimSuffix(n.Name(), ".md")
title, created := readTitleAndCreatedIn(bfs, hallRel+"/"+name, slug) full := filepath.Join(hallDir, n.Name())
title, created := readTitleAndCreated(full, slug)
out = append(out, noteEntry{Hall: hall, Slug: slug, Title: title, Created: created}) out = append(out, noteEntry{Hall: hall, Slug: slug, Title: title, Created: created})
} }
} }
@@ -124,40 +117,9 @@ func readTitleAndCreated(path, slug string) (string, string) {
return slug, "" return slug, ""
} }
defer func() { _ = f.Close() }() defer func() { _ = f.Close() }()
title, created := parseTitleAndCreated(f)
if created == "" {
if info, err := os.Stat(path); err == nil {
created = info.ModTime().UTC().Format("2006-01-02")
}
}
return titleOrSlug(title, slug), created
}
// readTitleAndCreatedIn is readTitleAndCreated through the brain root, title, created := "", ""
// regular files only (hyperguild#157). scanner := bufio.NewScanner(f)
func readTitleAndCreatedIn(bfs *brainfs.Brain, rel, slug string) (string, string) {
content, err := bfs.ReadRegular(rel)
if err != nil {
return slug, ""
}
title, created := parseTitleAndCreated(bytes.NewReader(content))
if created == "" {
if info, err := bfs.Lstat(rel); err == nil {
created = info.ModTime().UTC().Format("2006-01-02")
}
}
return titleOrSlug(title, slug), created
}
func titleOrSlug(title, slug string) string {
if title == "" {
return strings.ReplaceAll(slug, "-", " ")
}
return title
}
func parseTitleAndCreated(r io.Reader) (title, created string) {
scanner := bufio.NewScanner(r)
inFrontmatter := false inFrontmatter := false
for scanner.Scan() { for scanner.Scan() {
line := scanner.Text() line := scanner.Text()
@@ -187,5 +149,13 @@ func parseTitleAndCreated(r io.Reader) (title, created string) {
} }
} }
} }
if title == "" {
title = strings.ReplaceAll(slug, "-", " ")
}
if created == "" {
if info, err := os.Stat(path); err == nil {
created = info.ModTime().UTC().Format("2006-01-02")
}
}
return title, created return title, created
} }
-132
View File
@@ -1,132 +0,0 @@
package brain_test
import (
"os"
"path/filepath"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs/brainfstest"
)
func readIndex(t *testing.T, brainDir, wing string) string {
t.Helper()
b, _ := os.ReadFile(filepath.Join(brainDir, "wiki", wing, "_index.md"))
return string(b)
}
// hyperguild#157 slice 5: BuildWingIndex listed wiki/<wing>/<hall>/ with
// os.ReadDir and opened each note by path, so a symlinked note put the
// target's title into _index.md, and a FIFO hung the index build.
func TestBuildWingIndexRefusesSymlinkedNotesAndSurvivesSwapRaces(t *testing.T) {
b := t.TempDir()
facts := filepath.Join(b, "wiki", "w", "facts")
require.NoError(t, os.MkdirAll(facts, 0o755))
require.NoError(t, os.WriteFile(filepath.Join(facts, "ok.md"), []byte("---\ntitle: zqinside\n---\n"), 0o644))
require.NoError(t, os.Symlink(brainfstest.Outside(t), filepath.Join(facts, "leak.md")))
inBrain := brainfstest.WriteSentinelAt(t, filepath.Join(b, "raw", "s.md"))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "..", "raw", "s.md"), filepath.Join(facts, "inlink.md")))
require.NoError(t, brain.BuildWingIndex(b, "w"))
idx := readIndex(t, b, "w")
assert.Contains(t, idx, "zqinside", "positive control")
assert.NotContains(t, idx, brainfstest.Sentinel)
assert.NotContains(t, idx, "leak")
assert.NotContains(t, idx, "inlink")
p := filepath.Join(facts, "r.md")
read := func() string {
_ = brain.BuildWingIndex(b, "w")
return readIndex(t, b, "w")
}
brainfstest.AssertNoLeak(t, p, "---\ntitle: zqinside\n---\n", 1500*time.Millisecond, read)
brainfstest.AssertNoLeakFrom(t, p, "---\ntitle: zqinside\n---\n", inBrain, 1500*time.Millisecond, read)
brainfstest.AssertNoHang(t, p, "---\ntitle: zqinside\n---\n", 1500*time.Millisecond, 6*time.Second, func() { _ = read() })
}
// A symlinked hall is skipped while a real hall next to it is indexed; a
// symlinked wing is refused and nothing is written into its target.
func TestBuildWingIndexSkipsASymlinkedHallAndRefusesASymlinkedWing(t *testing.T) {
b := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(b, "raw", "facts"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(b, "raw", "facts", "r.md"), []byte("---\ntitle: zqrawonly\n---\n"), 0o644))
require.NoError(t, os.MkdirAll(filepath.Join(b, "wiki", "w", "decisions"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(b, "wiki", "w", "decisions", "d.md"), []byte("---\ntitle: zqrealhall\n---\n"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "facts"), filepath.Join(b, "wiki", "w", "facts")))
require.NoError(t, os.Symlink(filepath.Join("..", "raw"), filepath.Join(b, "wiki", "alias")))
require.NoError(t, brain.BuildWingIndex(b, "w"))
idx, err := os.ReadFile(filepath.Join(b, "wiki", "w", "_index.md"))
require.NoError(t, err, "the wing index is written")
assert.Contains(t, string(idx), "zqrealhall", "the real hall is indexed")
assert.NotContains(t, string(idx), "zqrawonly", "the symlinked hall is not")
assert.Error(t, brain.BuildWingIndex(b, "alias"))
_, err = os.Stat(filepath.Join(b, "raw", "_index.md"))
assert.True(t, os.IsNotExist(err), "an index was written into the symlinked wing's target")
}
// BuildAllWingIndexes indexes the real wings and skips a symlinked one.
func TestBuildAllWingIndexesSkipsASymlinkedWing(t *testing.T) {
b := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(b, "wiki", "w", "facts"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(b, "wiki", "w", "facts", "n.md"), []byte("---\ntitle: zqrealwing\n---\n"), 0o644))
require.NoError(t, os.MkdirAll(filepath.Join(b, "raw", "facts"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("..", "raw"), filepath.Join(b, "wiki", "alias")))
require.NoError(t, brain.BuildAllWingIndexes(b))
idx, err := os.ReadFile(filepath.Join(b, "wiki", "w", "_index.md"))
require.NoError(t, err)
assert.Contains(t, string(idx), "zqrealwing")
_, err = os.Stat(filepath.Join(b, "raw", "_index.md"))
assert.True(t, os.IsNotExist(err))
}
// A link to a symlinked note does not resolve: only regular files count as
// entries, so StripUnresolvedLinks unwraps it.
func TestStripUnresolvedLinksDoesNotCountASymlinkedNoteAsAnEntry(t *testing.T) {
b := t.TempDir()
facts := filepath.Join(b, "wiki", "w", "facts")
require.NoError(t, os.MkdirAll(facts, 0o755))
require.NoError(t, os.WriteFile(filepath.Join(facts, "real.md"), []byte("x"), 0o644))
brainfstest.WriteSentinelAt(t, filepath.Join(b, "raw", "s.md"))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "..", "raw", "s.md"), filepath.Join(facts, "ghost.md")))
out, unlinked, err := brain.StripUnresolvedLinks(b, "see [[real]] and [[ghost]]")
require.NoError(t, err)
assert.Contains(t, out, "[[real]]", "positive control")
assert.NotContains(t, out, "[[ghost]]")
assert.Contains(t, unlinked, "ghost")
}
// A linked tree (knowledge/ or wiki/) that is not a real directory is
// skipped, not an error: StripUnresolvedLinks runs before every brain_write,
// and an error there loses the learning.
func TestStripUnresolvedLinksSkipsALinkedTreeThatIsNotARealDirectory(t *testing.T) {
for name, mk := range map[string]func(t *testing.T, b string){
"symlink to an in-brain dir": func(t *testing.T, b string) {
require.NoError(t, os.MkdirAll(filepath.Join(b, "raw", "k"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("raw", "k"), filepath.Join(b, "knowledge")))
},
"dangling symlink": func(t *testing.T, b string) {
require.NoError(t, os.Symlink("nowhere", filepath.Join(b, "knowledge")))
},
"regular file": func(t *testing.T, b string) {
require.NoError(t, os.WriteFile(filepath.Join(b, "knowledge"), []byte("x"), 0o644))
},
} {
t.Run(name, func(t *testing.T) {
b := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(b, "wiki", "w", "facts"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(b, "wiki", "w", "facts", "real.md"), []byte("x"), 0o644))
mk(t, b)
out, _, err := brain.StripUnresolvedLinks(b, "learning [[real]] and [[gone]]")
require.NoError(t, err)
assert.Contains(t, out, "[[real]]", "links into wiki/ still resolve")
assert.NotContains(t, out, "[[gone]]")
})
}
}
+17 -32
View File
@@ -1,16 +1,12 @@
package brain package brain
import ( import (
"errors"
"fmt" "fmt"
"io/fs" "os"
"log/slog" "path/filepath"
"path"
"regexp" "regexp"
"sort" "sort"
"strings" "strings"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
) )
// wikilinkRE matches [[target]] and [[target|Display Text]]. // wikilinkRE matches [[target]] and [[target|Display Text]].
@@ -102,40 +98,29 @@ var linkedTrees = []string{"knowledge", "wiki"}
// one the audit will also resolve. // one the audit will also resolve.
func entryIndex(brainDir string) (map[string]struct{}, error) { func entryIndex(brainDir string) (map[string]struct{}, error) {
index := map[string]struct{}{} index := map[string]struct{}{}
// Walked through the brain root (hyperguild#157): only regular files
// count as entries, so a symlinked .md name cannot make a link resolve.
bfs, err := brainfs.Open(brainDir)
if err != nil {
return nil, fmt.Errorf("open brain: %w", err)
}
defer func() { _ = bfs.Close() }()
for _, dir := range linkedTrees { for _, dir := range linkedTrees {
if _, err := bfs.Lstat(dir); errors.Is(err, fs.ErrNotExist) { root := filepath.Join(brainDir, dir)
if _, err := os.Stat(root); os.IsNotExist(err) {
continue continue
} else if err != nil {
return nil, fmt.Errorf("stat %s: %w", dir, err)
} }
var walkErr error err := filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error {
err := bfs.WalkRegular(dir, func(rel string) error { if err != nil {
if !strings.HasSuffix(rel, ".md") { return err
}
if d.IsDir() || !strings.HasSuffix(path, ".md") {
return nil return nil
} }
index[strings.TrimSuffix(path.Base(rel), ".md")] = struct{}{} rel, err := filepath.Rel(brainDir, path)
if err != nil {
return err
}
rel = filepath.ToSlash(rel)
index[strings.TrimSuffix(filepath.Base(rel), ".md")] = struct{}{}
index[strings.TrimPrefix(strings.TrimSuffix(rel, ".md"), "wiki/")] = struct{}{} index[strings.TrimPrefix(strings.TrimSuffix(rel, ".md"), "wiki/")] = struct{}{}
return nil return nil
}, func(rel string, err error) {
// A linked tree that is not a real directory (a symlink, a
// file) is skipped, as filepath.WalkDir did: this runs before
// every brain_write, and an error here loses the learning.
if rel == dir && errors.Is(err, brainfs.ErrNotRegular) {
slog.Warn("link check: skipping a linked tree that is not a real directory", "dir", dir)
return
}
if walkErr == nil {
walkErr = fmt.Errorf("%s: %w", rel, err)
}
}) })
if err == nil {
err = walkErr
}
if err != nil { if err != nil {
return nil, fmt.Errorf("walk %s: %w", dir, err) return nil, fmt.Errorf("walk %s: %w", dir, err)
} }
+34 -84
View File
@@ -82,47 +82,42 @@ func (b *Brain) realDir(rel string) error {
return nil return nil
} }
// ReadDirRegular lists the regular files directly in a brain-relative // WriteRegular replaces the contents of an existing regular file at a
// directory, by name. Every component of relDir must be a real directory // brain-relative path. It never creates a file and never writes through a
// (ErrNotRegular otherwise); symlinks, FIFOs, devices and subdirectories in // final symlink: os.Root.OpenFile follows an in-root final symlink even with
// it are left out. // O_NOFOLLOW (verified on go1.27), so the path is Lstat-ed through the root
func (b *Brain) ReadDirRegular(relDir string) ([]string, error) { // first and the opened handle must be that same file (os.SameFile), which
relDir = filepath.ToSlash(relDir) // also catches a swap between the two. O_NONBLOCK keeps a FIFO from blocking
if err := b.realDir(relDir); err != nil { // the open. Intermediate in-brain symlinks are still followed (os.Root keeps
return nil, err // the write inside the brain; hyperguild#167).
} func (b *Brain) WriteRegular(rel string, data []byte) error {
ents, err := fs.ReadDir(b.root.FS(), relDir) rel = filepath.ToSlash(rel)
lfi, err := b.root.Lstat(rel)
if err != nil { if err != nil {
return nil, err return err
} }
var names []string if !lfi.Mode().IsRegular() {
for _, e := range ents { return fmt.Errorf("%s: %w", rel, ErrNotRegular)
if e.Type().IsRegular() {
names = append(names, e.Name())
}
} }
return names, nil f, err := b.root.OpenFile(rel, os.O_WRONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0)
}
// ReadDirDirs lists the real subdirectories directly in a brain-relative
// directory, by name. Every component of relDir must be a real directory
// (ErrNotRegular otherwise); symlinks to directories are left out.
func (b *Brain) ReadDirDirs(relDir string) ([]string, error) {
relDir = filepath.ToSlash(relDir)
if err := b.realDir(relDir); err != nil {
return nil, err
}
ents, err := fs.ReadDir(b.root.FS(), relDir)
if err != nil { if err != nil {
return nil, err return err
} }
var names []string defer func() { _ = f.Close() }()
for _, e := range ents { fi, err := f.Stat()
if e.Type().IsDir() { if err != nil {
names = append(names, e.Name()) return err
}
} }
return names, nil if !fi.Mode().IsRegular() || !os.SameFile(lfi, fi) {
return fmt.Errorf("%s: changed while opening: %w", rel, ErrNotRegular)
}
if err := f.Truncate(0); err != nil {
return err
}
if _, err := f.Write(data); err != nil {
return err
}
return f.Close()
} }
// ReadRegular reads a brain-relative path, refusing anything that is not a // ReadRegular reads a brain-relative path, refusing anything that is not a
@@ -135,7 +130,7 @@ func (b *Brain) ReadRegular(rel string) ([]byte, error) {
if !fi.Mode().IsRegular() { if !fi.Mode().IsRegular() {
return nil, fmt.Errorf("%s: %w", rel, ErrNotRegular) return nil, fmt.Errorf("%s: %w", rel, ErrNotRegular)
} }
return b.readOpenedMax(rel, -1, fi) return b.readOpened(rel)
} }
// readOpened opens rel non-blocking and reads it only if the OPENED handle is // readOpened opens rel non-blocking and reads it only if the OPENED handle is
@@ -149,7 +144,7 @@ func (b *Brain) ReadRegular(rel string) ([]byte, error) {
// cannot create one; fs.protected_hardlinks=1 and the pod's secrets are on a // cannot create one; fs.protected_hardlinks=1 and the pod's secrets are on a
// different filesystem from the brain volume, so a local caller cannot link // different filesystem from the brain volume, so a local caller cannot link
// them in. Refusing Nlink > 1 is the next step if either changes. // them in. Refusing Nlink > 1 is the next step if either changes.
func (b *Brain) readOpened(rel string) ([]byte, error) { return b.readOpenedMax(rel, -1, nil) } func (b *Brain) readOpened(rel string) ([]byte, error) { return b.readOpenedMax(rel, -1) }
// ReadRegularMax is ReadRegular with a size limit, checked on the opened // ReadRegularMax is ReadRegular with a size limit, checked on the opened
// handle before reading (max < 0 means no limit). // handle before reading (max < 0 means no limit).
@@ -161,14 +156,10 @@ func (b *Brain) ReadRegularMax(rel string, max int64) ([]byte, error) {
if !fi.Mode().IsRegular() { if !fi.Mode().IsRegular() {
return nil, fmt.Errorf("%s: %w", rel, ErrNotRegular) return nil, fmt.Errorf("%s: %w", rel, ErrNotRegular)
} }
return b.readOpenedMax(rel, max, fi) return b.readOpenedMax(rel, max)
} }
// readOpenedMax opens rel and reads it if the opened handle is a regular file func (b *Brain) readOpenedMax(rel string, max int64) ([]byte, error) {
// and, when lfi is set, the same file the caller Lstat-ed: os.Root follows an
// in-root final symlink even with O_NOFOLLOW, so a symlink to another brain
// file swapped in between the Lstat and the open is refused only here.
func (b *Brain) readOpenedMax(rel string, max int64, lfi fs.FileInfo) ([]byte, error) {
f, err := b.root.OpenFile(rel, os.O_RDONLY|syscall.O_NONBLOCK, 0) f, err := b.root.OpenFile(rel, os.O_RDONLY|syscall.O_NONBLOCK, 0)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -181,9 +172,6 @@ func (b *Brain) readOpenedMax(rel string, max int64, lfi fs.FileInfo) ([]byte, e
if !hfi.Mode().IsRegular() { if !hfi.Mode().IsRegular() {
return nil, fmt.Errorf("%s: %w", rel, ErrNotRegular) return nil, fmt.Errorf("%s: %w", rel, ErrNotRegular)
} }
if lfi != nil && !os.SameFile(lfi, hfi) {
return nil, fmt.Errorf("%s: changed while opening: %w", rel, ErrNotRegular)
}
if max >= 0 && hfi.Size() > max { if max >= 0 && hfi.Size() > max {
return nil, fmt.Errorf("%s is %d bytes (limit %d): %w", rel, hfi.Size(), max, ErrTooLarge) return nil, fmt.Errorf("%s is %d bytes (limit %d): %w", rel, hfi.Size(), max, ErrTooLarge)
} }
@@ -249,41 +237,3 @@ func (b *Brain) WalkRegular(relDir string, fn func(rel string) error, onErr func
return fn(p) return fn(p)
}) })
} }
// WriteRegular replaces the contents of an existing regular file at a
// brain-relative path. It never creates a file and never writes through a
// final symlink: os.Root.OpenFile follows an in-root final symlink even with
// O_NOFOLLOW (verified on go1.27), so the path is Lstat-ed through the root
// first and the opened handle must be that same file (os.SameFile), which
// also catches a swap between the two. O_NONBLOCK keeps a FIFO from blocking
// the open. Intermediate in-brain symlinks are still followed (os.Root keeps
// the write inside the brain; hyperguild#167).
func (b *Brain) WriteRegular(rel string, data []byte) error {
rel = filepath.ToSlash(rel)
lfi, err := b.root.Lstat(rel)
if err != nil {
return err
}
if !lfi.Mode().IsRegular() {
return fmt.Errorf("%s: %w", rel, ErrNotRegular)
}
f, err := b.root.OpenFile(rel, os.O_WRONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0)
if err != nil {
return err
}
defer func() { _ = f.Close() }()
fi, err := f.Stat()
if err != nil {
return err
}
if !fi.Mode().IsRegular() || !os.SameFile(lfi, fi) {
return fmt.Errorf("%s: changed while opening: %w", rel, ErrNotRegular)
}
if err := f.Truncate(0); err != nil {
return err
}
if _, err := f.Write(data); err != nil {
return err
}
return f.Close()
}
@@ -6,8 +6,6 @@ import (
"io/fs" "io/fs"
"os" "os"
"path/filepath" "path/filepath"
"sync"
"sync/atomic"
"syscall" "syscall"
"testing" "testing"
"time" "time"
@@ -192,77 +190,6 @@ func TestWalkRegularReportsASymlinkedIntermediateComponentOfTheWalkRoot(t *testi
assert.ErrorIs(t, errs[0], ErrNotRegular) assert.ErrorIs(t, errs[0], ErrNotRegular)
} }
func TestReadDirRegularListsOnlyRegularFilesAndRefusesSymlinkedComponents(t *testing.T) {
dir := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(dir, "a", "b", "sub"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(dir, "a", "b", "n.md"), []byte("n"), 0o644))
require.NoError(t, os.WriteFile(filepath.Join(dir, "other.md"), []byte("o"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "other.md"), filepath.Join(dir, "a", "b", "l.md")))
require.NoError(t, syscall.Mkfifo(filepath.Join(dir, "a", "b", "f.md"), 0o644))
require.NoError(t, os.Symlink("a", filepath.Join(dir, "alias")))
require.NoError(t, os.MkdirAll(filepath.Join(dir, "x"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("..", "a"), filepath.Join(dir, "x", "mid")))
b, err := Open(dir)
require.NoError(t, err)
defer func() { _ = b.Close() }()
names, err := b.ReadDirRegular("a/b")
require.NoError(t, err)
assert.Equal(t, []string{"n.md"}, names)
for _, rel := range []string{"alias/b", "x/mid/b"} {
_, err := b.ReadDirRegular(rel)
assert.ErrorIs(t, err, ErrNotRegular, rel)
}
}
// A final component swapped for a symlink to another regular file in the
// brain between the Lstat and the open: os.Root follows in-root links, so
// only comparing the two (os.SameFile) refuses it.
func TestReadRegularNeverReturnsAnInBrainFileSwappedInBehindASymlink(t *testing.T) {
dir := t.TempDir()
require.NoError(t, os.WriteFile(filepath.Join(dir, "s.md"), []byte("zqsamefilesentinel"), 0o644))
p := filepath.Join(dir, "r.md")
require.NoError(t, os.WriteFile(p, []byte("inside"), 0o644))
b, err := Open(dir)
require.NoError(t, err)
defer func() { _ = b.Close() }()
var stop atomic.Bool
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
tmp := p + ".swap"
for i := 0; !stop.Load(); i++ {
_ = os.Remove(tmp)
if i%2 == 0 {
_ = os.WriteFile(tmp, []byte("inside"), 0o644)
} else {
_ = os.Symlink("s.md", tmp)
}
_ = os.Rename(tmp, p)
}
}()
leaks, ok := 0, 0
for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); {
got, err := b.ReadRegular("r.md")
if err == nil {
ok++
}
if string(got) == "zqsamefilesentinel" {
leaks++
}
got, _ = b.ReadRegularMax("r.md", 1<<20)
if string(got) == "zqsamefilesentinel" {
leaks++
}
}
stop.Store(true)
wg.Wait()
assert.Zero(t, leaks)
assert.Positive(t, ok, "positive control: the regular file is read")
}
func TestWriteRegularRefusesFinalSymlinksFIFOsEscapesAndMissingFilesAndRewritesRegularFiles(t *testing.T) { func TestWriteRegularRefusesFinalSymlinksFIFOsEscapesAndMissingFilesAndRewritesRegularFiles(t *testing.T) {
dir := t.TempDir() dir := t.TempDir()
outside := filepath.Join(t.TempDir(), "o.md") outside := filepath.Join(t.TempDir(), "o.md")
@@ -308,21 +235,3 @@ func TestWriteRegularRefusesFinalSymlinksFIFOsEscapesAndMissingFilesAndRewritesR
got, _ = os.ReadFile(filepath.Join(dir, "n.md")) got, _ = os.ReadFile(filepath.Join(dir, "n.md"))
assert.Equal(t, "new", string(got), "truncated and rewritten") assert.Equal(t, "new", string(got), "truncated and rewritten")
} }
func TestReadDirDirsListsOnlyRealSubdirsAndRefusesSymlinkedComponents(t *testing.T) {
dir := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(dir, "a", "real"), 0o755))
require.NoError(t, os.MkdirAll(filepath.Join(dir, "other"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("..", "other"), filepath.Join(dir, "a", "linked")))
require.NoError(t, os.WriteFile(filepath.Join(dir, "a", "f.md"), []byte("f"), 0o644))
require.NoError(t, os.Symlink("a", filepath.Join(dir, "alias")))
b, err := Open(dir)
require.NoError(t, err)
defer func() { _ = b.Close() }()
names, err := b.ReadDirDirs("a")
require.NoError(t, err)
assert.Equal(t, []string{"real"}, names)
_, err = b.ReadDirDirs("alias")
assert.ErrorIs(t, err, ErrNotRegular)
}
@@ -7,7 +7,6 @@
package brainfstest package brainfstest
import ( import (
"fmt"
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
@@ -21,47 +20,29 @@ import (
// Sentinel marks the outside file's contents. // Sentinel marks the outside file's contents.
const Sentinel = "zqbrainfssentinelonefiftyseven" const Sentinel = "zqbrainfssentinelonefiftyseven"
// outsideContent carries Sentinel in every field a reader might extract: the
// frontmatter title and aliases, a heading, a wikilink and the body.
const outsideContent = "---\ntitle: " + Sentinel + "\naliases:\n - " + Sentinel + "\n---\n# " + Sentinel +
"\n\n" + Sentinel + " [[" + Sentinel + "]]\n"
// Outside writes a file outside any brain dir containing Sentinel. // Outside writes a file outside any brain dir containing Sentinel.
func Outside(t testing.TB) string { func Outside(t *testing.T) string {
t.Helper() t.Helper()
p := filepath.Join(t.TempDir(), "secret.md") p := filepath.Join(t.TempDir(), "secret.md")
if err := os.WriteFile(p, []byte(outsideContent), 0o644); err != nil { if err := os.WriteFile(p, []byte("# secret\n\n"+Sentinel), 0o644); err != nil {
t.Fatal(err) t.Fatal(err)
} }
return p return p
} }
var swapSeq atomic.Int64
// StartSwap renames, in a loop, a regular note (content inside) and either // StartSwap renames, in a loop, a regular note (content inside) and either
// a symlink to outside or a FIFO into path, until the test ends. FIFOs get a // a symlink to outside or a FIFO into path, until the test ends. FIFOs get a
// writer held open until the swapper exits, so a reader that skips the // writer held open until the swapper exits, so a reader that skips the
// handle check parks forever instead of reading EOF. The swapper is stopped // handle check parks forever instead of reading EOF. The swapper is stopped
// and awaited before t.TempDir cleanup. It returns the number of symlinks // and awaited before t.TempDir cleanup.
// or FIFOs actually renamed into path so far, so a caller can tell a race func StartSwap(t *testing.T, path, inside, outside string, fifo bool) {
// that ran from one whose every swap failed (hyperguild#177).
func StartSwap(t testing.TB, path, inside, outside string, fifo bool) *atomic.Int64 {
t.Helper() t.Helper()
var swaps atomic.Int64
var stop atomic.Bool var stop atomic.Bool
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Add(1) wg.Add(1)
go func() { go func() {
defer wg.Done() defer wg.Done()
// A name of its own: swappers on one path share a test, and one tmp := path + ".swap"
// opening another's FIFO for write would block forever.
tmp := fmt.Sprintf("%s.swap%d", path, swapSeq.Add(1))
// A relative link: os.Root refuses every absolute link on its own, so
// an absolute in-brain target would test nothing.
link := outside
if rel, err := filepath.Rel(filepath.Dir(path), outside); err == nil {
link = rel
}
var writers []*os.File var writers []*os.File
defer func() { defer func() {
for _, w := range writers { for _, w := range writers {
@@ -70,85 +51,36 @@ func StartSwap(t testing.TB, path, inside, outside string, fifo bool) *atomic.In
}() }()
for i := 0; !stop.Load(); i++ { for i := 0; !stop.Load(); i++ {
_ = os.Remove(tmp) _ = os.Remove(tmp)
var made error
switch { switch {
case i%2 == 0: case i%2 == 0:
_ = os.WriteFile(tmp, []byte(inside), 0o644) _ = os.WriteFile(tmp, []byte(inside), 0o644)
case fifo: case fifo:
if made = syscall.Mkfifo(tmp, 0o644); made == nil { if err := syscall.Mkfifo(tmp, 0o644); err == nil {
if w, err := os.OpenFile(tmp, os.O_RDWR, 0); err == nil { if w, err := os.OpenFile(tmp, os.O_RDWR, 0); err == nil {
writers = append(writers, w) writers = append(writers, w)
} }
} }
default: default:
made = os.Symlink(link, tmp) _ = os.Symlink(outside, tmp)
}
if err := os.Rename(tmp, path); err == nil && made == nil && i%2 == 1 {
swaps.Add(1)
} }
_ = os.Rename(tmp, path)
} }
}() }()
t.Cleanup(func() { stop.Store(true); wg.Wait() }) t.Cleanup(func() { stop.Store(true); wg.Wait() })
return &swaps
}
// WriteSentinelAt writes a file carrying Sentinel at p (creating its
// directory), for an in-brain symlink target.
func WriteSentinelAt(t *testing.T, p string) string {
t.Helper()
if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(p, []byte(outsideContent), 0o644); err != nil {
t.Fatal(err)
}
return p
} }
// AssertNoLeak loops read for d while a symlink swapper runs on path, and // AssertNoLeak loops read for d while a symlink swapper runs on path, and
// fails if any output contains Sentinel. The symlink points outside the // fails if any output contains Sentinel.
// brain; os.Root refuses that on its own, so pair it with AssertNoLeakFrom
// and an in-brain target.
func AssertNoLeak(t *testing.T, path, inside string, d time.Duration, read func() string) { func AssertNoLeak(t *testing.T, path, inside string, d time.Duration, read func() string) {
t.Helper() t.Helper()
AssertNoLeakFrom(t, path, inside, Outside(t), d, read) StartSwap(t, path, inside, Outside(t), false)
} leaks, iters := 0, 0
// InsideMarker must appear in the inside note's content and in read's
// output whenever read saw that note: the floor below counts such reads.
const InsideMarker = "inside"
// AssertNoLeakFrom is AssertNoLeak with the symlink pointing at target, e.g.
// a file elsewhere in the brain written with WriteSentinelAt.
//
// Beyond "no Sentinel in any output" it sets two floors, because without
// them the assertion passes when nothing was tested (hyperguild#177): at
// least one symlink must have been swapped into path, and at least one read
// must have returned InsideMarker (a reader that always errors returns
// nothing, and so never leaks).
func AssertNoLeakFrom(t testing.TB, path, inside, target string, d time.Duration, read func() string) {
t.Helper()
if !strings.Contains(inside, InsideMarker) {
t.Fatalf("inside content must contain %q", InsideMarker)
}
swaps := StartSwap(t, path, inside, target, false)
leaks, iters, insideReads := 0, 0, 0
for deadline := time.Now().Add(d); time.Now().Before(deadline); iters++ { for deadline := time.Now().Add(d); time.Now().Before(deadline); iters++ {
out := read() leaks += strings.Count(read(), Sentinel)
leaks += strings.Count(out, Sentinel)
if strings.Contains(out, InsideMarker) {
insideReads++
}
} }
if leaks > 0 { if leaks > 0 {
t.Fatalf("outside file leaked %d times in %d iterations", leaks, iters) t.Fatalf("outside file leaked %d times in %d iterations", leaks, iters)
} }
if n := swaps.Load(); n == 0 {
t.Fatalf("no symlink was swapped into %s in %d iterations: the race did not run", path, iters)
}
if insideReads == 0 {
t.Fatalf("no read in %d iterations returned the inside note (%q): the reader read nothing", iters, InsideMarker)
}
} }
// AssertNoHang loops read for d while a FIFO swapper (held writers) runs on // AssertNoHang loops read for d while a FIFO swapper (held writers) runs on
@@ -1,79 +0,0 @@
package brainfstest
import (
"fmt"
"os"
"path/filepath"
"runtime"
"syscall"
"testing"
"time"
)
// recordingTB stands in for the test under AssertNoLeakFrom so a fatal
// there can be observed instead of failing this test. Fatalf ends the
// calling goroutine the way testing.T does.
type recordingTB struct {
testing.TB
failed bool
msg string
}
func (r *recordingTB) Fatalf(format string, args ...any) {
r.failed, r.msg = true, fmt.Sprintf(format, args...)
runtime.Goexit()
}
func runAssert(t *testing.T, path, target string, read func() string) *recordingTB {
t.Helper()
r := &recordingTB{TB: t}
done := make(chan struct{})
go func() {
defer close(done)
AssertNoLeakFrom(r, path, "inside", target, 200*time.Millisecond, read)
}()
<-done
return r
}
// hyperguild#177: with every swap failing, no leak can occur, so the leak
// assertion passed without testing anything.
func TestAssertNoLeakFromFailsWhenNoSymlinkSwapSucceeds(t *testing.T) {
path := filepath.Join(t.TempDir(), "missing-dir", "r.md")
r := runAssert(t, path, Outside(t), func() string { return "inside" })
if !r.failed {
t.Fatal("AssertNoLeakFrom passed with zero successful symlink swaps")
}
}
// hyperguild#177: a reader that always errors returns nothing, so it never
// leaks either.
func TestAssertNoLeakFromFailsWhenNoReadReturnsTheInsideNote(t *testing.T) {
path := filepath.Join(t.TempDir(), "r.md")
r := runAssert(t, path, Outside(t), func() string { return "" })
if !r.failed {
t.Fatal("AssertNoLeakFrom passed with zero reads returning the inside note")
}
}
// Positive control: a reader that refuses symlinks and reads regular files
// passes, so the floors do not fail a working reader.
func TestAssertNoLeakFromPassesAReaderThatRefusesSymlinks(t *testing.T) {
path := filepath.Join(t.TempDir(), "r.md")
read := func() string {
f, err := os.OpenFile(path, os.O_RDONLY|noFollow, 0)
if err != nil {
return ""
}
defer func() { _ = f.Close() }()
b := make([]byte, 64)
n, _ := f.Read(b)
return string(b[:n])
}
r := runAssert(t, path, Outside(t), read)
if r.failed {
t.Fatalf("a symlink-refusing reader failed: %s", r.msg)
}
}
const noFollow = syscall.O_NOFOLLOW
@@ -16,10 +16,9 @@
package classification package classification
import ( import (
"errors"
"fmt" "fmt"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "os"
"io/fs" "path/filepath"
"strings" "strings"
"gopkg.in/yaml.v3" "gopkg.in/yaml.v3"
@@ -111,20 +110,9 @@ type rawConfig struct {
func Load(brainDir string) (*Config, error) { func Load(brainDir string) (*Config, error) {
cfg := &Config{wings: map[string]Level{}, repos: map[string]Level{}} cfg := &Config{wings: map[string]Level{}, repos: map[string]Level{}}
// Read through the brain root, regular files only (hyperguild#157): a data, err := os.ReadFile(filepath.Join(brainDir, "classification.yaml"))
// symlinked or special file is not trusted as the config, which is a hard
// error like a malformed one.
bfs, err := brainfs.Open(brainDir)
if errors.Is(err, fs.ErrNotExist) {
return cfg, nil
}
if err != nil { if err != nil {
return nil, fmt.Errorf("open brain: %w", err) if os.IsNotExist(err) {
}
defer func() { _ = bfs.Close() }()
data, err := bfs.ReadRegular("classification.yaml")
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
return cfg, nil return cfg, nil
} }
return nil, fmt.Errorf("read classification.yaml: %w", err) return nil, fmt.Errorf("read classification.yaml: %w", err)
@@ -1,46 +0,0 @@
package classification
import (
"os"
"path/filepath"
"syscall"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// hyperguild#157 slice 6: Load read classification.yaml with os.ReadFile. It
// sets the I1 sovereignty levels; a symlinked file must not be trusted as the
// config (a hard error, as for a malformed file), and a FIFO must not hang
// startup.
func TestLoadRefusesASymlinkedOrFIFOClassificationFile(t *testing.T) {
ok := t.TempDir()
require.NoError(t, os.WriteFile(filepath.Join(ok, "classification.yaml"), []byte("wings:\n w: confidential\n"), 0o644))
_, err := Load(ok)
require.NoError(t, err, "positive control")
linked := t.TempDir()
require.NoError(t, os.WriteFile(filepath.Join(linked, "real.yaml"), []byte("wings:\n w: public\n"), 0o644))
require.NoError(t, os.Symlink("real.yaml", filepath.Join(linked, "classification.yaml")))
_, err = Load(linked)
assert.Error(t, err, "an in-brain symlink is not trusted as the config")
fifo := t.TempDir()
require.NoError(t, syscall.Mkfifo(filepath.Join(fifo, "classification.yaml"), 0o644))
w, err := os.OpenFile(filepath.Join(fifo, "classification.yaml"), os.O_RDWR, 0)
require.NoError(t, err)
defer func() { _ = w.Close() }()
done := make(chan error, 1)
go func() { _, err := Load(fifo); done <- err }()
select {
case err := <-done:
assert.Error(t, err)
case <-time.After(3 * time.Second):
t.Fatal("Load hung on a FIFO")
}
_, err = Load(t.TempDir())
assert.NoError(t, err, "an absent file is still not an error")
}
+21 -43
View File
@@ -12,14 +12,10 @@ package graphsync
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"io/fs" "os"
"path"
"path/filepath" "path/filepath"
"strings"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"github.com/mathiasbq/hyperguild/ingestion/internal/graph" "github.com/mathiasbq/hyperguild/ingestion/internal/graph"
"github.com/mathiasbq/hyperguild/ingestion/internal/graphstore" "github.com/mathiasbq/hyperguild/ingestion/internal/graphstore"
) )
@@ -50,25 +46,8 @@ func IndexDoc(ctx context.Context, store Store, brainDir, relPath string) error
if relPath == "" { if relPath == "" {
return nil return nil
} }
// Callers pass caller-supplied paths (brain_tunnel, brain_update): index abs := filepath.Join(brainDir, filepath.FromSlash(relPath))
// only a clean path under wiki/ or knowledge/. content, err := os.ReadFile(abs)
if path.Clean(relPath) != relPath || !filepath.IsLocal(relPath) ||
(!strings.HasPrefix(relPath, "wiki/") && !strings.HasPrefix(relPath, "knowledge/")) {
return fmt.Errorf("index %q: not a clean wiki/ or knowledge/ path", relPath)
}
bfs, err := brainfs.Open(brainDir)
if err != nil {
return fmt.Errorf("open brain: %w", err)
}
defer func() { _ = bfs.Close() }()
return indexDoc(ctx, store, bfs, relPath)
}
// indexDoc reads relPath through the brain root, regular files only
// (hyperguild#157): the brain is a Syncthing folder, and os.ReadFile
// followed a synced symlink into the graph and blocked on a FIFO.
func indexDoc(ctx context.Context, store Store, bfs *brainfs.Brain, relPath string) error {
content, err := bfs.ReadRegular(relPath)
if err != nil { if err != nil {
return fmt.Errorf("read %q: %w", relPath, err) return fmt.Errorf("read %q: %w", relPath, err)
} }
@@ -98,34 +77,33 @@ func BackfillFromBrainDir(ctx context.Context, store Store, brainDir string) (in
if store == nil { if store == nil {
return 0, nil return 0, nil
} }
bfs, err := brainfs.Open(brainDir) roots := []string{"wiki", "knowledge"}
if err != nil { for _, root := range roots {
return 0, fmt.Errorf("open brain: %w", err) base := filepath.Join(brainDir, root)
} if _, err := os.Stat(base); os.IsNotExist(err) {
defer func() { _ = bfs.Close() }()
for _, root := range []string{"wiki", "knowledge"} {
if _, err := bfs.Lstat(root); errors.Is(err, fs.ErrNotExist) {
continue continue
} }
// Any walk error, including a root that is a symlink, aborts. err := filepath.WalkDir(base, func(path string, d os.DirEntry, walkErr error) error {
var walkErr error if walkErr != nil {
err := bfs.WalkRegular(root, func(rel string) error { return walkErr
if path.Ext(rel) != ".md" { }
if d.IsDir() {
return nil return nil
} }
if err := indexDoc(ctx, store, bfs, rel); err != nil { if filepath.Ext(path) != ".md" {
return nil
}
rel, relErr := filepath.Rel(brainDir, path)
if relErr != nil {
return fmt.Errorf("rel %q: %w", path, relErr)
}
rel = filepath.ToSlash(rel)
if err := IndexDoc(ctx, store, brainDir, rel); err != nil {
return fmt.Errorf("index %q: %w", rel, err) return fmt.Errorf("index %q: %w", rel, err)
} }
indexed++ indexed++
return nil return nil
}, func(rel string, err error) {
if walkErr == nil {
walkErr = fmt.Errorf("%s: %w", rel, err)
}
}) })
if err == nil {
err = walkErr
}
if err != nil { if err != nil {
return indexed, fmt.Errorf("walk %s: %w", root, err) return indexed, fmt.Errorf("walk %s: %w", root, err)
} }
@@ -1,92 +0,0 @@
package graphsync
import (
"context"
"fmt"
"os"
"path/filepath"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs/brainfstest"
)
func (f *fakeStore) dump() string {
f.mu.Lock()
defer f.mu.Unlock()
return fmt.Sprintf("%+v %+v", f.upserts, f.replaces)
}
// hyperguild#157 slice 4: IndexDoc and BackfillFromBrainDir read notes with
// os.ReadFile, so a symlinked note put the target's title and wikilinks into
// the graph, and a FIFO hung the read.
func TestIndexDocRefusesSymlinksAndSurvivesSwapRaces(t *testing.T) {
brain := t.TempDir()
facts := filepath.Join(brain, "wiki", "w", "facts")
require.NoError(t, os.MkdirAll(facts, 0o755))
require.NoError(t, os.WriteFile(filepath.Join(facts, "ok.md"), []byte("---\ntitle: zqinside\n---\nbody [[other]]\n"), 0o644))
require.NoError(t, os.Symlink(brainfstest.Outside(t), filepath.Join(facts, "leak.md")))
store := newFakeStore()
require.NoError(t, IndexDoc(context.Background(), store, brain, "wiki/w/facts/ok.md"))
assert.Contains(t, store.dump(), "zqinside", "positive control: a regular note is indexed")
assert.Error(t, IndexDoc(context.Background(), store, brain, "wiki/w/facts/leak.md"))
// A link to a file elsewhere in the brain: os.Root allows it, only the
// regular-file check refuses it.
inBrain := brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md"))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "..", "raw", "s.md"), filepath.Join(facts, "inlink.md")))
assert.Error(t, IndexDoc(context.Background(), store, brain, "wiki/w/facts/inlink.md"))
assert.NotContains(t, store.dump(), brainfstest.Sentinel)
p := filepath.Join(facts, "r.md")
read := func() string {
s := newFakeStore()
_ = IndexDoc(context.Background(), s, brain, "wiki/w/facts/r.md")
return s.dump()
}
brainfstest.AssertNoLeak(t, p, "---\ntitle: zqinside\n---\n", 1500*time.Millisecond, read)
brainfstest.AssertNoLeakFrom(t, p, "---\ntitle: zqinside\n---\n", inBrain, 1500*time.Millisecond, read)
brainfstest.AssertNoHang(t, p, "---\ntitle: zqinside\n---\n", 1500*time.Millisecond, 6*time.Second, func() { _ = read() })
}
func TestBackfillSkipsSymlinkedNotesAndAbortsOnASymlinkedRoot(t *testing.T) {
brain := t.TempDir()
facts := filepath.Join(brain, "wiki", "w", "facts")
require.NoError(t, os.MkdirAll(facts, 0o755))
require.NoError(t, os.WriteFile(filepath.Join(facts, "ok.md"), []byte("---\ntitle: zqinside\n---\n"), 0o644))
require.NoError(t, os.Symlink(brainfstest.Outside(t), filepath.Join(facts, "leak.md")))
require.NoError(t, os.MkdirAll(filepath.Join(brain, "raw"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(brain, "raw", "r.md"), []byte("---\ntitle: zqrawonly\n---\n"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "..", "raw", "r.md"), filepath.Join(facts, "inlink.md")))
require.NoError(t, os.Symlink(".", filepath.Join(brain, "knowledge")))
store := newFakeStore()
_, err := BackfillFromBrainDir(context.Background(), store, brain)
assert.Error(t, err, "a symlinked root aborts the backfill")
assert.NotContains(t, store.dump(), brainfstest.Sentinel)
assert.NotContains(t, store.dump(), "zqrawonly")
require.NoError(t, os.Remove(filepath.Join(brain, "knowledge")))
store = newFakeStore()
n, err := BackfillFromBrainDir(context.Background(), store, brain)
require.NoError(t, err)
assert.Equal(t, 1, n, "only the regular note is indexed")
assert.Contains(t, store.dump(), "zqinside")
assert.NotContains(t, store.dump(), brainfstest.Sentinel)
}
// IndexDoc gets caller-supplied paths (brain_tunnel, brain_update): only a
// clean path under wiki/ or knowledge/ is indexed.
func TestIndexDocRefusesPathsOutsideWikiAndKnowledgeAndUncleanPaths(t *testing.T) {
brain := t.TempDir()
brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "secret.md"))
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki", "w"), 0o755))
for _, p := range []string{"raw/secret.md", "wiki/w/../../raw/secret.md", "./raw/secret.md", "/raw/secret.md"} {
store := newFakeStore()
assert.Error(t, IndexDoc(context.Background(), store, brain, p), p)
assert.NotContains(t, store.dump(), brainfstest.Sentinel, p)
}
}
+8 -17
View File
@@ -3,10 +3,7 @@ package pipeline
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"io/fs"
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
@@ -26,16 +23,10 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
return 0, fmt.Errorf("load inventory: %w", err) return 0, fmt.Errorf("load inventory: %w", err)
} }
// Listed and read through the brain root, regular files only sourcesDir := filepath.Join(brainDir, "wiki", "sources")
// (hyperguild#157). entries, err := os.ReadDir(sourcesDir)
bfs, err := brainfs.Open(brainDir)
if err != nil { if err != nil {
return 0, fmt.Errorf("open brain: %w", err) if os.IsNotExist(err) {
}
defer func() { _ = bfs.Close() }()
names, err := bfs.ReadDirRegular("wiki/sources")
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
return 0, nil return 0, nil
} }
return 0, fmt.Errorf("read sources dir: %w", err) return 0, fmt.Errorf("read sources dir: %w", err)
@@ -46,20 +37,20 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
// get both refs merged before a single write. // get both refs merged before a single write.
pending := make(map[string]wiki.Page) pending := make(map[string]wiki.Page)
for _, name := range names { for _, e := range entries {
if ctx.Err() != nil { if ctx.Err() != nil {
return 0, ctx.Err() return 0, ctx.Err()
} }
if !strings.HasSuffix(name, ".md") { if e.IsDir() || !strings.HasSuffix(e.Name(), ".md") {
continue continue
} }
b, err := bfs.ReadRegular("wiki/sources/" + name) b, err := os.ReadFile(filepath.Join(sourcesDir, e.Name()))
if err != nil { if err != nil {
continue continue
} }
sourceContent := string(b) sourceContent := string(b)
sourceSlug := strings.TrimSuffix(name, ".md") sourceSlug := strings.TrimSuffix(e.Name(), ".md")
sourceTitle := extractTitle(sourceContent) sourceTitle := extractTitle(sourceContent)
if sourceTitle == "" { if sourceTitle == "" {
sourceTitle = sourceSlug sourceTitle = sourceSlug
@@ -79,7 +70,7 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
// Start from already-accumulated version if we've seen this page. // Start from already-accumulated version if we've seen this page.
page, seen := pending[relPath] page, seen := pending[relPath]
if !seen { if !seen {
raw, err := bfs.ReadRegular(relPath) raw, err := os.ReadFile(filepath.Join(brainDir, filepath.FromSlash(relPath)))
if err != nil { if err != nil {
continue continue
} }
+6 -22
View File
@@ -5,10 +5,7 @@ import (
"context" "context"
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"errors"
"fmt" "fmt"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"io/fs"
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
@@ -132,7 +129,7 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br
if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil { if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
return Result{}, fmt.Errorf("mkdir for %s: %w", page.Path, err) return Result{}, fmt.Errorf("mkdir for %s: %w", page.Path, err)
} }
content, conflicts, err := reviseOrCreate(brainDir, page) content, conflicts, err := reviseOrCreate(dest, page)
if err != nil { if err != nil {
return Result{}, err return Result{}, err
} }
@@ -191,27 +188,14 @@ func hashBody(content string) string {
// readSourcePage returns the full on-disk content of the source page at // readSourcePage returns the full on-disk content of the source page at
// wiki/sources/<sourceSlug>.md, or "" when it doesn't exist yet. // wiki/sources/<sourceSlug>.md, or "" when it doesn't exist yet.
// It reads through the brain root, regular files only (hyperguild#157): the
// content goes into the LLM prompt.
func readSourcePage(brainDir, sourceSlug string) string { func readSourcePage(brainDir, sourceSlug string) string {
raw, err := readBrainFile(brainDir, "wiki/sources/"+sourceSlug+".md") raw, err := os.ReadFile(filepath.Join(brainDir, "wiki", "sources", sourceSlug+".md"))
if err != nil { if err != nil {
return "" return ""
} }
return string(raw) return string(raw)
} }
// readBrainFile reads one brain-relative file through the brain root,
// regular files only (hyperguild#157).
func readBrainFile(brainDir, rel string) ([]byte, error) {
bfs, err := brainfs.Open(brainDir)
if err != nil {
return nil, err
}
defer func() { _ = bfs.Close() }()
return bfs.ReadRegular(rel)
}
// storedSourceHash reads source_hash from the source page a previous ingest // storedSourceHash reads source_hash from the source page a previous ingest
// wrote, or "" when there is no such page or no such field. // wrote, or "" when there is no such page or no such field.
// //
@@ -246,10 +230,10 @@ func storedSourceHash(brainDir, sourceSlug string) string {
// its first-ingest date on a revised page. Left deliberately — the alternative // its first-ingest date on a revised page. Left deliberately — the alternative
// is this function editing frontmatter it does not own, and no lint reads that // is this function editing frontmatter it does not own, and no lint reads that
// field today. // field today.
func reviseOrCreate(brainDir string, page wiki.Page) (string, []string, error) { func reviseOrCreate(dest string, page wiki.Page) (string, []string, error) {
existing, err := readBrainFile(brainDir, page.Path) existing, err := os.ReadFile(dest)
switch { switch {
case errors.Is(err, fs.ErrNotExist): case os.IsNotExist(err):
return page.Content, nil, nil return page.Content, nil, nil
case err != nil: case err != nil:
return "", nil, fmt.Errorf("read existing %s: %w", page.Path, err) return "", nil, fmt.Errorf("read existing %s: %w", page.Path, err)
@@ -283,7 +267,7 @@ See brain/schema.md for the full schema.
` `
func loadSchema(brainDir string) string { func loadSchema(brainDir string) string {
b, err := readBrainFile(brainDir, "schema.md") b, err := os.ReadFile(filepath.Join(brainDir, "schema.md"))
if err != nil { if err != nil {
return defaultSchema return defaultSchema
} }
+3 -1
View File
@@ -2,6 +2,7 @@
package pipeline package pipeline
import ( import (
"os"
"path/filepath" "path/filepath"
"regexp" "regexp"
"strings" "strings"
@@ -52,7 +53,8 @@ func injectSourceRefs(pages []wiki.Page, inventory map[wiki.PageType][]wiki.Entr
if !ok { if !ok {
continue continue
} }
b, err := readBrainFile(brainDir, "wiki/"+string(pt)+"/"+slug+".md") diskPath := filepath.Join(brainDir, "wiki", string(pt), slug+".md")
b, err := os.ReadFile(diskPath)
if err != nil { if err != nil {
continue continue
} }
-167
View File
@@ -1,167 +0,0 @@
package pipeline
import (
"context"
"fmt"
"io"
"os"
"path/filepath"
"syscall"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs/brainfstest"
"github.com/mathiasbq/hyperguild/ingestion/internal/wiki"
)
// hyperguild#157 slice 6: the pipeline read the source page, schema.md, the
// pages it revises and the pages it adds source refs to with os.ReadFile, so
// a symlinked file's content reached the LLM prompt and the written pages,
// and a FIFO hung the ingest.
func mkdirs(t *testing.T, brain string, dirs ...string) {
t.Helper()
for _, d := range dirs {
require.NoError(t, os.MkdirAll(filepath.Join(brain, filepath.FromSlash(d)), 0o755))
}
}
func TestReadSourcePageAndSchemaRefuseSymlinksAndSurviveSwapRaces(t *testing.T) {
brain := t.TempDir()
mkdirs(t, brain, "wiki/sources")
inBrain := brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md"))
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "ok.md"), []byte("zqinside source"), 0o644))
require.NoError(t, os.WriteFile(filepath.Join(brain, "schema.md"), []byte("zqinside schema"), 0o644))
assert.Equal(t, "zqinside source", readSourcePage(brain, "ok"), "positive control")
assert.Equal(t, "zqinside schema", loadSchema(brain), "positive control")
require.NoError(t, os.Symlink(brainfstest.Outside(t), filepath.Join(brain, "wiki", "sources", "out.md")))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "s.md"), filepath.Join(brain, "wiki", "sources", "in.md")))
assert.NotContains(t, readSourcePage(brain, "out"), brainfstest.Sentinel)
assert.NotContains(t, readSourcePage(brain, "in"), brainfstest.Sentinel)
assert.NotContains(t, storedSourceHash(brain, "in"), brainfstest.Sentinel)
src := filepath.Join(brain, "wiki", "sources", "r.md")
read := func() string { return readSourcePage(brain, "r") }
brainfstest.AssertNoLeak(t, src, "zqinside", 1500*time.Millisecond, read)
brainfstest.AssertNoLeakFrom(t, src, "zqinside", inBrain, 1500*time.Millisecond, read)
brainfstest.AssertNoHang(t, src, "zqinside", 1500*time.Millisecond, 6*time.Second, func() { _ = read() })
schema := filepath.Join(brain, "schema.md")
rs := func() string { return loadSchema(brain) }
brainfstest.AssertNoLeakFrom(t, schema, "zqinside schema", inBrain, 1500*time.Millisecond, rs)
brainfstest.AssertNoHang(t, schema, "zqinside schema", 1500*time.Millisecond, 6*time.Second, func() { _ = rs() })
}
func TestReviseOrCreateRefusesASymlinkedExistingPage(t *testing.T) {
brain := t.TempDir()
mkdirs(t, brain, "wiki/concepts")
brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md"))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "s.md"), filepath.Join(brain, "wiki", "concepts", "c.md")))
page := wiki.Page{Path: "wiki/concepts/c.md", Content: "---\ntitle: C\n---\nnew\n"}
out, _, err := reviseOrCreate(brain, page)
assert.Error(t, err)
assert.NotContains(t, out, brainfstest.Sentinel)
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "concepts", "d.md"), []byte("---\ntitle: D\n---\nzqold\n"), 0o644))
out, _, err = reviseOrCreate(brain, wiki.Page{Path: "wiki/concepts/d.md", Content: "---\ntitle: D\n---\nnew\n"})
require.NoError(t, err)
assert.Contains(t, out, "zqold", "positive control: the existing page is read")
out, _, err = reviseOrCreate(brain, wiki.Page{Path: "wiki/concepts/new.md", Content: "fresh"})
require.NoError(t, err)
assert.Equal(t, "fresh", out, "positive control: a missing page is created")
}
func TestInjectSourceRefsDoesNotReadASymlinkedPage(t *testing.T) {
brain := t.TempDir()
mkdirs(t, brain, "wiki/concepts")
brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md"))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "s.md"), filepath.Join(brain, "wiki", "concepts", "c.md")))
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "concepts", "d.md"), []byte("zqinside d"), 0o644))
inv := map[wiki.PageType][]wiki.Entry{wiki.PageTypeConcept: {
{Slug: "c", Type: wiki.PageTypeConcept}, {Slug: "d", Type: wiki.PageTypeConcept},
}}
src := wiki.Page{Path: "wiki/sources/src.md", Content: "---\ntitle: Src\n---\nsee [[c|C]] and [[d|D]]\n"}
out := injectSourceRefs([]wiki.Page{src}, inv, brain)
got := fmt.Sprintf("%+v", out)
assert.Contains(t, got, "zqinside d", "positive control")
assert.NotContains(t, got, brainfstest.Sentinel)
}
func TestBackfillRefsSkipsSymlinkedSourcesAndPages(t *testing.T) {
brain := t.TempDir()
mkdirs(t, brain, "wiki/sources", "wiki/concepts", "raw")
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "concepts", "real.md"), []byte("---\ntitle: R\n---\nbody\n"), 0o644))
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "ok.md"), []byte("---\ntitle: zqinside\n---\n[[real|R]]\n"), 0o644))
// A source elsewhere in the brain whose title is the sentinel and which
// links the real concept with the [[slug|title]] form the pipeline follows.
require.NoError(t, os.WriteFile(filepath.Join(brain, "raw", "src.md"),
[]byte("---\ntitle: "+brainfstest.Sentinel+"\n---\n[[real|R]]\n"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "src.md"), filepath.Join(brain, "wiki", "sources", "in.md")))
// A concept page linked by the regular source that is itself a symlink.
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "ok2.md"), []byte("---\ntitle: Two\n---\n[[lnk|L]]\n"), 0o644))
brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "page.md"))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "page.md"), filepath.Join(brain, "wiki", "concepts", "lnk.md")))
_, err := BackfillRefs(context.Background(), brain)
require.NoError(t, err)
b, _ := os.ReadFile(filepath.Join(brain, "wiki", "concepts", "real.md"))
assert.Contains(t, string(b), "zqinside", "positive control: the regular source's ref is added")
assert.NotContains(t, string(b), brainfstest.Sentinel, "a symlinked source's title was written as a ref")
raw, _ := os.ReadFile(filepath.Join(brain, "raw", "page.md"))
assert.NotContains(t, string(raw), "[[ok2", "the symlinked concept page's target was rewritten")
}
// A source swapped between a regular file and a symlink to an in-brain source
// whose title is the sentinel: the read refuses the swapped-in link, so the
// sentinel never lands in the concept page as a ref.
func TestBackfillRefsDoesNotWriteARefFromASourceSwappedInBehindASymlink(t *testing.T) {
brain := t.TempDir()
mkdirs(t, brain, "wiki/sources", "wiki/concepts", "raw")
realPage := filepath.Join(brain, "wiki", "concepts", "real.md")
target := filepath.Join(brain, "raw", "src.md")
require.NoError(t, os.WriteFile(target, []byte("---\ntitle: "+brainfstest.Sentinel+"\n---\n[[real|R]]\n"), 0o644))
src := filepath.Join(brain, "wiki", "sources", "r.md")
inside := "---\ntitle: zqinside\n---\n[[real|R]]\n"
read := func() string {
_ = os.WriteFile(realPage, []byte("---\ntitle: R\n---\nbody\n"), 0o644)
_, _ = BackfillRefs(context.Background(), brain)
b, _ := os.ReadFile(realPage)
return string(b)
}
require.NoError(t, os.WriteFile(src, []byte(inside), 0o644))
require.Contains(t, read(), "zqinside", "positive control")
brainfstest.AssertNoLeakFrom(t, src, inside, target, 1500*time.Millisecond, read)
}
// A concept page that is regular when the inventory lists it, then swapped
// for a symlink to an in-brain file before BackfillRefs reads it: only the
// page read's own check refuses it (a static symlink never gets this far,
// because LoadInventory does not list it).
func TestBackfillRefsDoesNotCopyAPageSwappedInBehindASymlink(t *testing.T) {
brain := t.TempDir()
mkdirs(t, brain, "wiki/sources", "wiki/concepts", "raw")
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "s.md"), []byte("---\ntitle: S\n---\n[[c|C]]\n"), 0o644))
page := filepath.Join(brain, "wiki", "concepts", "c.md")
target := filepath.Join(brain, "raw", "secret.md")
inside := "---\ntitle: C\n---\nzqinside\n"
read := func() string {
// The write path still follows a swapped-in link (#158) and can
// overwrite the target, so restore it each round.
_ = os.WriteFile(target, []byte("---\ntitle: X\n---\n"+brainfstest.Sentinel+"\n"), 0o644)
_, _ = BackfillRefs(context.Background(), brain)
f, err := os.OpenFile(page, os.O_RDONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0)
if err != nil {
return ""
}
defer func() { _ = f.Close() }()
b, _ := io.ReadAll(f)
return string(b)
}
require.NoError(t, os.WriteFile(page, []byte(inside), 0o644))
require.Contains(t, read(), "[[s|S]]", "positive control: the regular page gets the source ref")
brainfstest.AssertNoLeakFrom(t, page, inside, target, 3*time.Second, read)
}
+9 -26
View File
@@ -93,20 +93,16 @@ func TestQueryDoesNotFollowASymlinkedWingUnderAWingAndHallRoot(t *testing.T) {
// bypass of readOpened hung Query after 5 iterations). // bypass of readOpened hung Query after 5 iterations).
// startSwap runs swapInto until the test ends and waits for it before // startSwap runs swapInto until the test ends and waits for it before
// t.TempDir's cleanup (hyperguild#162). // t.TempDir's cleanup (hyperguild#162).
// It returns the count of symlinks or FIFOs actually renamed into path func startSwap(t *testing.T, path, outside string, fifo bool) {
// (hyperguild#177: a swapper whose every swap fails tests nothing).
func startSwap(t *testing.T, path, outside string, fifo bool) *atomic.Int64 {
t.Helper() t.Helper()
var stop atomic.Bool var stop atomic.Bool
var swaps atomic.Int64
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Add(1) wg.Add(1)
go func() { defer wg.Done(); swapInto(path, outside, fifo, &stop, &swaps) }() go func() { defer wg.Done(); swapInto(path, outside, fifo, &stop) }()
t.Cleanup(func() { stop.Store(true); wg.Wait() }) t.Cleanup(func() { stop.Store(true); wg.Wait() })
return &swaps
} }
func swapInto(path, outside string, fifo bool, stop *atomic.Bool, swaps *atomic.Int64) { func swapInto(path, outside string, fifo bool, stop *atomic.Bool) {
tmp := path + ".swap" tmp := path + ".swap"
// Writers stay open until the swapper exits, so a reader parked on a // Writers stay open until the swapper exits, so a reader parked on a
// FIFO stays parked (closing one would hand the reader EOF and hide the // FIFO stays parked (closing one would hand the reader EOF and hide the
@@ -119,46 +115,33 @@ func swapInto(path, outside string, fifo bool, stop *atomic.Bool, swaps *atomic.
}() }()
for i := 0; !stop.Load(); i++ { for i := 0; !stop.Load(); i++ {
_ = os.Remove(tmp) _ = os.Remove(tmp)
var made error
switch { switch {
case i%2 == 0: case i%2 == 0:
_ = os.WriteFile(tmp, []byte("# r\n\nzqinside "+"zqracenote"), 0o644) _ = os.WriteFile(tmp, []byte("# r\n\nzqinside "+"zqracenote"), 0o644)
case fifo: case fifo:
if made = syscall.Mkfifo(tmp, 0o644); made == nil { if err := syscall.Mkfifo(tmp, 0o644); err == nil {
if w, err := os.OpenFile(tmp, os.O_RDWR, 0); err == nil { if w, err := os.OpenFile(tmp, os.O_RDWR, 0); err == nil {
writers = append(writers, w) writers = append(writers, w)
} }
} }
default: default:
made = os.Symlink(outside, tmp) _ = os.Symlink(outside, tmp)
}
if err := os.Rename(tmp, path); err == nil && made == nil && i%2 == 1 {
swaps.Add(1)
} }
_ = os.Rename(tmp, path)
} }
} }
func TestQueryLeaksNothingUnderASymlinkSwapRace(t *testing.T) { func TestQueryLeaksNothingUnderASymlinkSwapRace(t *testing.T) {
brain, facts, outside := searchBrain(t) brain, facts, outside := searchBrain(t)
swaps := startSwap(t, filepath.Join(facts, "r.md"), outside, false) startSwap(t, filepath.Join(facts, "r.md"), outside, false)
leaks, iters, insideReads := 0, 0, 0 leaks, iters := 0, 0
for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); iters++ { for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); iters++ {
rs, err := Query(brain, QueryOptions{Query: sentinel157, Limit: 10}) rs, _ := Query(brain, QueryOptions{Query: sentinel157, Limit: 10})
require.NoError(t, err)
leaks += leaked(rs) leaks += leaked(rs)
// Positive control (hyperguild#177): a Query that always errors
// returns nothing and so never leaks.
rs, err = Query(brain, QueryOptions{Query: "zqracenote", Limit: 10})
require.NoError(t, err)
if len(rs) > 0 {
insideReads++
}
} }
if leaks > 0 { if leaks > 0 {
t.Fatalf("outside file leaked %d times in %d iterations", leaks, iters) t.Fatalf("outside file leaked %d times in %d iterations", leaks, iters)
} }
require.Positive(t, swaps.Load(), "no symlink was swapped in: the race did not run")
require.Positive(t, insideReads, "no Query returned the inside note")
} }
func TestQueryDoesNotHangUnderAFIFOSwapRace(t *testing.T) { func TestQueryDoesNotHangUnderAFIFOSwapRace(t *testing.T) {
-43
View File
@@ -1,12 +1,9 @@
package session package session
import ( import (
"errors"
"os" "os"
"path/filepath" "path/filepath"
"syscall"
"testing" "testing"
"time"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
@@ -50,43 +47,3 @@ func TestReadRefusesSessionIDsThatAreNotPlainNames(t *testing.T) {
_, err := Read(filepath.Join(root, "sessions"), "../x") _, err := Read(filepath.Join(root, "sessions"), "../x")
assert.Error(t, err) assert.Error(t, err)
} }
// A name that starts plainly and then climbs out with ".." is exactly what the
// ban on "/" in the id's tail stops: filepath.Join cleans "a/../../x" to a
// path beside the sessions dir without needing sessions/a to exist. The
// assertion is on the refusal itself (ErrInvalidID), not on any error.
func TestAppendRefusesAnIDThatClimbsOutAfterAPlainPrefix(t *testing.T) {
root := t.TempDir()
dir := filepath.Join(root, "sessions")
require.NoError(t, os.MkdirAll(dir, 0o755))
for _, id := range []string{"a/../../escape", "a/../../../escape", "a/../b"} {
err := Append(dir, id, Entry{SessionID: id})
assert.ErrorIs(t, err, ErrInvalidID, "%q", id)
}
_, err := os.Lstat(filepath.Join(root, "escape.jsonl"))
assert.True(t, errors.Is(err, os.ErrNotExist), "nothing created beside the sessions dir")
_, err = os.Lstat(filepath.Join(dir, "b.jsonl"))
assert.True(t, errors.Is(err, os.ErrNotExist), "no in-dir alias created either")
}
// Append does no Lstat type check, so O_NONBLOCK is the only thing that keeps
// a FIFO named <id>.jsonl with no reader from blocking session_log forever.
func TestAppendReturnsPromptlyOnAFIFOLogWithNoReader(t *testing.T) {
dir := filepath.Join(t.TempDir(), "sessions")
require.NoError(t, os.MkdirAll(dir, 0o755))
fifo := filepath.Join(dir, "s.jsonl")
require.NoError(t, syscall.Mkfifo(fifo, 0o644))
done := make(chan error, 1)
go func() { done <- Append(dir, "s", Entry{SessionID: "s"}) }()
select {
case err := <-done:
assert.Error(t, err, "a FIFO with no reader is refused, not written")
case <-time.After(3 * time.Second):
// Unblock the stuck open so the goroutine and TempDir cleanup finish.
if r, err := os.OpenFile(fifo, os.O_RDONLY|syscall.O_NONBLOCK, 0); err == nil {
defer r.Close() //nolint:errcheck
}
<-done
t.Fatal("Append blocked on a FIFO log with no reader")
}
}
+1 -4
View File
@@ -49,12 +49,9 @@ type Attempt struct {
// write (hyperguild#169). // write (hyperguild#169).
var idPattern = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9._-]{0,127}$`) var idPattern = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9._-]{0,127}$`)
// ErrInvalidID is returned when a session id is not a plain file name.
var ErrInvalidID = errors.New("session id is not a plain name")
func checkID(sessionID string) error { func checkID(sessionID string) error {
if !idPattern.MatchString(sessionID) { if !idPattern.MatchString(sessionID) {
return fmt.Errorf("%w: %q", ErrInvalidID, sessionID) return fmt.Errorf("session id %q is not a plain name", sessionID)
} }
return nil return nil
} }
@@ -0,0 +1,75 @@
// Package startupretry retries a startup dependency that may not be serving
// yet, instead of exiting on its first failure.
//
// It exists for the cold-boot ordering race (#83): on a cluster restart the
// ingestion pod can reach OIDC discovery before dex is answering, gets a 502
// from ingress, and dies — after having already connected postgres, backfilled
// the graph and ingested session batches, all of which the restart redoes.
// Nothing is misconfigured in that scenario; the pod simply started first.
//
// The budget is bounded on purpose. Retrying forever would turn a genuine
// misconfiguration (wrong issuer URL, bad audience) into a process that looks
// healthy and never serves, which is harder to diagnose than a crash loop.
package startupretry
import (
"context"
"fmt"
"time"
)
// Policy bounds the retry: at most Attempts calls, with the delay doubling
// from Base and capped at Max.
type Policy struct {
Attempts int // total calls, including the first; < 1 means 1
Base time.Duration // delay before the second attempt
Max time.Duration // ceiling for the doubling delay
}
// Do calls fn until it succeeds, the attempt budget is spent, or ctx ends.
//
// The returned error wraps the last failure, so callers keep the diagnosis
// they would have had without the retry.
func Do[T any](ctx context.Context, p Policy, fn func(context.Context) (T, error)) (T, error) {
var zero T
attempts := p.Attempts
if attempts < 1 {
attempts = 1
}
delay := p.Base
var lastErr error
for i := range attempts {
if err := ctx.Err(); err != nil {
return zero, fmt.Errorf("startupretry: %w", err)
}
v, err := fn(ctx)
if err == nil {
return v, nil
}
lastErr = err
if i == attempts-1 {
break
}
if delay <= 0 {
delay = time.Second
}
if p.Max > 0 && delay > p.Max {
delay = p.Max
}
select {
case <-ctx.Done():
return zero, fmt.Errorf("startupretry: %w", ctx.Err())
case <-time.After(delay):
}
delay *= 2
}
return zero, fmt.Errorf("startupretry: gave up after %d attempts: %w", attempts, lastErr)
}
@@ -0,0 +1,142 @@
// ingestion/internal/startupretry/startupretry_test.go
package startupretry
import (
"context"
"errors"
"testing"
"time"
)
func fastPolicy(attempts int) Policy {
return Policy{Attempts: attempts, Base: time.Millisecond, Max: 2 * time.Millisecond}
}
func TestDo_ReturnsImmediatelyOnFirstSuccess(t *testing.T) {
calls := 0
got, err := Do(context.Background(), fastPolicy(5), func(context.Context) (string, error) {
calls++
return "ready", nil
})
if err != nil {
t.Fatalf("Do: %v", err)
}
if got != "ready" {
t.Errorf("got %q, want %q", got, "ready")
}
if calls != 1 {
t.Errorf("calls = %d, want 1", calls)
}
}
func TestDo_SurvivesTransientFailuresThenSucceeds(t *testing.T) {
// The observed cold-boot race: dex is not serving yet, ingress answers
// 502 for the first few attempts, then discovery works.
calls := 0
got, err := Do(context.Background(), fastPolicy(5), func(context.Context) (string, error) {
calls++
if calls < 4 {
return "", errors.New("oidc discovery: status 502")
}
return "ready", nil
})
if err != nil {
t.Fatalf("Do returned %v — a transient 502 must not be fatal", err)
}
if got != "ready" {
t.Errorf("got %q, want %q", got, "ready")
}
if calls != 4 {
t.Errorf("calls = %d, want 4", calls)
}
}
func TestDo_GivesUpAfterTheAttemptBudgetAndReportsTheLastError(t *testing.T) {
// A genuinely broken issuer must still fail — retrying forever would turn
// a config error into a silent hang, which is worse than crash-looping.
last := errors.New("oidc discovery: status 404")
calls := 0
_, err := Do(context.Background(), fastPolicy(3), func(context.Context) (string, error) {
calls++
return "", last
})
if !errors.Is(err, last) {
t.Fatalf("err = %v, want it to wrap the last failure", err)
}
if calls != 3 {
t.Errorf("calls = %d, want 3 (the attempt budget)", calls)
}
}
func TestDo_StopsPromptlyWhenTheContextIsCancelled(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
calls := 0
done := make(chan struct{})
var err error
go func() {
defer close(done)
_, err = Do(ctx, Policy{Attempts: 100, Base: 50 * time.Millisecond, Max: time.Second},
func(context.Context) (string, error) {
calls++
if calls == 1 {
cancel()
}
return "", errors.New("still down")
})
}()
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("Do did not return after context cancellation")
}
if !errors.Is(err, context.Canceled) {
t.Errorf("err = %v, want context.Canceled", err)
}
}
func TestDo_BacksOffAndIsCappedByMax(t *testing.T) {
var delays []time.Duration
p := Policy{Attempts: 5, Base: 10 * time.Millisecond, Max: 25 * time.Millisecond}
last := time.Now()
_, _ = Do(context.Background(), p, func(context.Context) (string, error) {
now := time.Now()
delays = append(delays, now.Sub(last))
last = now
return "", errors.New("down")
})
// delays[0] is the first call (no wait); the rest are the sleeps.
if len(delays) != 5 {
t.Fatalf("attempts = %d, want 5", len(delays))
}
if delays[1] >= delays[2] {
t.Errorf("backoff is not growing: %v then %v", delays[1], delays[2])
}
for i, d := range delays[1:] {
if d > 2*p.Max {
t.Errorf("delay %d = %v, exceeds the Max cap %v with slack", i+1, d, p.Max)
}
}
}
func TestPolicy_ZeroValueStillMakesOneAttempt(t *testing.T) {
// A caller that forgets to configure the policy must not silently get a
// no-op that returns a nil validator and a nil error.
calls := 0
_, err := Do(context.Background(), Policy{}, func(context.Context) (string, error) {
calls++
return "", errors.New("down")
})
if calls != 1 {
t.Errorf("calls = %d, want 1", calls)
}
if err == nil {
t.Error("err = nil, want the failure surfaced")
}
}
@@ -18,14 +18,9 @@ import (
"sort" "sort"
"strconv" "strconv"
"strings" "strings"
"sync"
"sync/atomic"
"time" "time"
) )
// shutdownFlushTimeout bounds the final flush when Watch stops.
var shutdownFlushTimeout = 5 * time.Second
// DefaultInterval is used when Watch is given a non-positive interval. // DefaultInterval is used when Watch is given a non-positive interval.
const DefaultInterval = 30 * time.Second const DefaultInterval = 30 * time.Second
@@ -138,32 +133,13 @@ func Scan(brainDir string) (Result, error) {
// queue until Notify succeeds (see alertQueue): an outage delays them, and // queue until Notify succeeds (see alertQueue): an outage delays them, and
// any that do not fit the bounded queue, or are undeliverable, are counted in // any that do not fit the bounded queue, or are undeliverable, are counted in
// a summary that is never dropped, so every incident is delivered or counted. // a summary that is never dropped, so every incident is delivered or counted.
//
// Delivery runs in its own goroutine (hyperguild#164): a hung notifier, with
// a full queue, blocked one flush for several notify timeouts, and while the
// scan waited on it a planted link stayed readable for several intervals.
// The scan hands alerts over without blocking; any that do not fit are
// counted in the summary. When ctx ends, Watch returns after a final flush
// of what was handed over (bounded by shutdownFlushTimeout); alerts that
// flush cannot deliver are counted in one error log line instead. That
// shutdown path only runs for a caller that cancels ctx: cmd/server passes
// context.Background(), so in production the flush on handoff is what keeps
// alerts from waiting on a restart.
func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) { func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
if interval <= 0 { if interval <= 0 {
interval = DefaultInterval interval = DefaultInterval
} }
t := time.NewTicker(interval) t := time.NewTicker(interval)
defer t.Stop() defer t.Stop()
alerts := make(chan string, maxPending) q := &alertQueue{}
var overflow atomic.Int64
var wg sync.WaitGroup
wg.Add(1)
go func() { defer wg.Done(); deliver(ctx, interval, n, alerts, &overflow) }()
// The scan is the only sender: closing alerts tells deliver to drain and
// flush once more, and Watch returns only after it has.
defer func() { close(alerts); wg.Wait() }()
q := alertSender{alerts: alerts, overflow: &overflow}
reported := "" reported := ""
for { for {
select { select {
@@ -192,76 +168,7 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
". Readers may follow symlinks out of the brain; fix them.") ". Readers may follow symlinks out of the brain; fix them.")
} }
} }
} q.flush(ctx, n)
}
// alertSender hands alerts to deliver without ever blocking the scan; an
// alert that does not fit is counted in overflow, which deliver adds to the
// summary.
type alertSender struct {
alerts chan<- string
overflow *atomic.Int64
}
func (s alertSender) push(msg string) {
select {
case s.alerts <- msg:
default:
s.overflow.Add(1)
}
}
// deliver owns the alertQueue. It flushes as soon as an alert is handed
// over, and again every interval while any are pending (outage retries).
// When alerts is closed (Watch is stopping) it drains it and flushes once
// more, bounded by shutdownFlushTimeout, so an alert handed over just before
// shutdown is not lost (PR #180 review). Whatever that flush cannot deliver
// is logged as one error with its counts, never dropped silently.
func deliver(ctx context.Context, interval time.Duration, n Notifier, alerts <-chan string, overflow *atomic.Int64) {
t := time.NewTicker(interval)
defer t.Stop()
q := &alertQueue{}
// drain moves everything handed over into the queue; it reports false
// once alerts is closed.
drain := func() bool {
defer func() { q.suppressed += int(overflow.Swap(0)) }()
for {
select {
case msg, ok := <-alerts:
if !ok {
return false
}
q.push(msg)
default:
return true
}
}
}
for {
select {
case msg, ok := <-alerts:
if ok {
q.push(msg)
ok = drain()
}
if !ok {
q.suppressed += int(overflow.Swap(0))
fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownFlushTimeout)
q.flush(fctx, n)
cancel()
if len(q.items) > 0 || q.suppressed > 0 {
slog.Error("symlink tripwire alerts not delivered at shutdown",
"undelivered", len(q.items), "suppressed", q.suppressed)
}
return
}
q.flush(ctx, n)
case <-t.C:
if !drain() {
continue // the closed channel is handled by the case above
}
q.flush(ctx, n)
}
} }
} }
@@ -1,12 +1,10 @@
package symlinkguard package symlinkguard
import ( import (
"bytes"
"context" "context"
"errors" "errors"
"fmt" "fmt"
"os" "os"
"log/slog"
"path/filepath" "path/filepath"
"strconv" "strconv"
"strings" "strings"
@@ -347,19 +345,13 @@ func TestWatchDropsAMessageThatNeverSendsSoLaterAlertsFlow(t *testing.T) {
}, 3*time.Second, 10*time.Millisecond, "a later alert is delivered despite a poison message") }, 3*time.Second, 10*time.Millisecond, "a later alert is delivered despite a poison message")
} }
// toggleNotifier fails while down is set. Like the real audit.NtfyNotifier // toggleNotifier fails while down is set.
// (http.NewRequestWithContext), it also fails, without recording, when its
// ctx is already done (PR #180 round 2: notifiers that ignored ctx let a
// final flush on the cancelled ctx pass).
type toggleNotifier struct { type toggleNotifier struct {
recNotifier recNotifier
down atomic.Bool down atomic.Bool
} }
func (r *toggleNotifier) Notify(ctx context.Context, m string) error { func (r *toggleNotifier) Notify(ctx context.Context, m string) error {
if err := ctx.Err(); err != nil {
return err
}
if r.down.Load() { if r.down.Load() {
return errors.New("ntfy down") return errors.New("ntfy down")
} }
@@ -504,271 +496,3 @@ func TestWatchDropsAPoisonMessageAndCountsIt(t *testing.T) {
return false return false
}, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary") }, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary")
} }
// hungNotifier answers every Notify only after latency (or when ctx ends),
// with an outage error: an ntfy that accepts connections but never responds
// until its timeout.
type hungNotifier struct{ latency time.Duration }
func (h hungNotifier) Notify(ctx context.Context, _ string) error {
select {
case <-time.After(h.latency):
case <-ctx.Done():
}
return errors.New("ntfy timeout")
}
func waitGone(t *testing.T, link string) {
t.Helper()
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 10*time.Second, time.Millisecond)
}
// hyperguild#164: alerts were flushed in the scan goroutine, so with a hung
// ntfy and a full queue one flush (6 x notify timeout) delayed the next scan.
// The notifier here is slower than the interval, as in production (10 s
// timeout vs 30 s interval with 6 messages), so the alert handoff is full:
// a scan that waited on it would keep a link for seconds.
func TestWatchLinkLifetimeStaysNearOneIntervalWithANotifierSlowerThanTheInterval(t *testing.T) {
const interval = 100 * time.Millisecond
brain := t.TempDir()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go Watch(ctx, brain, interval, hungNotifier{latency: 400 * time.Millisecond})
for i := 0; i < 2*maxPending+3; i++ { // overfill queue and handoff
link := filepath.Join(brain, "fill"+strconv.Itoa(i))
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
}
var worst time.Duration
for i := 0; i < 6; i++ {
link := filepath.Join(brain, "probe"+strconv.Itoa(i))
require.NoError(t, os.Symlink(t.TempDir(), link))
planted := time.Now()
waitGone(t, link)
worst = max(worst, time.Since(planted))
}
assert.LessOrEqual(t, worst, interval+interval/2, "worst link lifetime %v with a hung notifier", worst)
}
// Review of PR #180: delivery waited for its own tick, about one interval
// after the removal. An alert is now flushed as soon as it is handed over.
func TestWatchDeliversARemovalAlertWellWithinOneInterval(t *testing.T) {
const interval = 400 * time.Millisecond
brain := t.TempDir()
n := &recNotifier{}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go Watch(ctx, brain, interval, n)
link := filepath.Join(brain, "prompt")
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
removed := time.Now()
require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, time.Millisecond)
assert.Less(t, time.Since(removed), interval/4)
}
// Review of PR #180: on ctx end, deliver returned without draining what the
// scan had handed over, so a removal just before shutdown was neither
// delivered nor counted. Watch now drains and flushes before returning.
func TestWatchDeliversAnAlertHandedOverJustBeforeShutdown(t *testing.T) {
for i := 0; i < 20; i++ {
brain := t.TempDir()
n := &recNotifier{}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }()
link := filepath.Join(brain, "late")
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
cancel()
<-done
require.Equal(t, 1, n.count(), "iteration %d: the removal alert was lost at shutdown", i)
}
}
// blockingNotifier holds every Notify until its ctx ends, tracking calls in
// flight.
type blockingNotifier struct{ inFlight atomic.Int32 }
func (b *blockingNotifier) Notify(ctx context.Context, _ string) error {
b.inFlight.Add(1)
defer b.inFlight.Add(-1)
<-ctx.Done()
return ctx.Err()
}
// Watch does not return while its delivery goroutine is still notifying.
func TestWatchReturnsOnlyAfterDeliveryHasStopped(t *testing.T) {
prev := shutdownFlushTimeout
shutdownFlushTimeout = 50 * time.Millisecond
t.Cleanup(func() { shutdownFlushTimeout = prev })
brain := t.TempDir()
n := &blockingNotifier{}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }()
link := filepath.Join(brain, "x")
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
require.Eventually(t, func() bool { return n.inFlight.Load() == 1 }, 2*time.Second, time.Millisecond)
cancel()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("Watch did not return after cancel")
}
assert.Zero(t, n.inFlight.Load(), "Notify still running after Watch returned")
}
// slowToggleNotifier fails slowly (a hung ntfy) while down, then records.
type slowToggleNotifier struct {
recNotifier
down atomic.Bool
latency time.Duration
}
func (r *slowToggleNotifier) Notify(ctx context.Context, m string) error {
if r.down.Load() {
select {
case <-time.After(r.latency):
case <-ctx.Done():
}
return errors.New("ntfy timeout")
}
return r.recNotifier.Notify(ctx, m)
}
// Review of PR #180: with a slow notifier the scan's handoff overflows, and
// those alerts exist only in the overflow counter. Every removal is still
// delivered or counted in the summary.
func TestWatchCountsHandoffOverflowWithASlowNotifier(t *testing.T) {
brain := t.TempDir()
n := &slowToggleNotifier{latency: 30 * time.Millisecond}
n.down.Store(true)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go Watch(ctx, brain, 2*time.Millisecond, n)
const plants = 40
for i := 0; i < plants; i++ {
link := filepath.Join(brain, "ov"+strconv.Itoa(i))
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
}
n.down.Store(false)
delivered, counted := 0, 0
require.Eventually(t, func() bool {
n.mu.Lock()
defer n.mu.Unlock()
delivered, counted = 0, 0
for _, m := range n.msgs {
if strings.Contains(m, "removed 1 symlink(s)") {
delivered++
}
if i := strings.Index(m, "): "); i >= 0 && strings.Contains(m, "suppressed") {
var c int
_, _ = fmt.Sscanf(m[i+3:], "%d", &c)
counted += c
}
}
return delivered+counted == plants
}, 5*time.Second, 10*time.Millisecond, "delivered %d + counted %d of %d", delivered, counted, plants)
}
// An alert still queued when Watch stops (its send failed, the notifier has
// since recovered, the retry tick has not come) is flushed once more before
// Watch returns, on a ctx that is not the cancelled one: toggleNotifier fails
// on a done ctx, as the real ntfy notifier does.
func TestWatchFinalFlushSendsAQueuedAlertOnACtxThatOutlivesTheCancel(t *testing.T) {
brain := t.TempDir()
n := &toggleNotifier{}
n.down.Store(true)
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { Watch(ctx, brain, 300*time.Millisecond, n); close(done) }()
link := filepath.Join(brain, "queued")
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
time.Sleep(20 * time.Millisecond) // let the immediate send fail
n.down.Store(false)
cancel()
<-done
assert.Equal(t, 1, n.count(), "the queued alert was not flushed at shutdown")
}
// lockedBuffer is a goroutine-safe log sink: Watch goroutines of earlier
// tests may still log while one test captures.
type lockedBuffer struct {
mu sync.Mutex
buf bytes.Buffer
}
func (b *lockedBuffer) Write(p []byte) (int, error) {
b.mu.Lock()
defer b.mu.Unlock()
return b.buf.Write(p)
}
func (b *lockedBuffer) String() string { b.mu.Lock(); defer b.mu.Unlock(); return b.buf.String() }
func captureLog(t *testing.T) *lockedBuffer {
t.Helper()
b := &lockedBuffer{}
prev := slog.Default()
slog.SetDefault(slog.New(slog.NewTextHandler(b, nil)))
t.Cleanup(func() { slog.SetDefault(prev) })
return b
}
// PR #180 round 2: when the final flush fails (ntfy hung at shutdown) the
// undelivered alerts vanished with only "will retry" warnings. Watch now logs
// one error with both counts.
func TestWatchLogsUndeliveredAndSuppressedCountsWhenTheFinalFlushFails(t *testing.T) {
prev := shutdownFlushTimeout
shutdownFlushTimeout = 50 * time.Millisecond
t.Cleanup(func() { shutdownFlushTimeout = prev })
logs := captureLog(t)
brain := t.TempDir()
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { Watch(ctx, brain, 5*time.Millisecond, hungNotifier{latency: time.Hour}); close(done) }()
link := filepath.Join(brain, "lost")
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
cancel()
<-done
out := logs.String()
assert.Contains(t, out, "level=ERROR")
assert.Contains(t, out, `msg="symlink tripwire alerts not delivered at shutdown"`)
assert.Contains(t, out, "undelivered=1")
assert.Contains(t, out, "suppressed=0")
}
// deliver, given a handoff that already holds alerts and is closed, moves them
// into the queue and runs the final flush, even though the ctx is cancelled:
// the drain-on-close path, without depending on when the close lands.
func TestDeliverFlushesAlertsDrainedFromAClosedHandoffOnACancelledCtx(t *testing.T) {
alerts := make(chan string, 3)
alerts <- "a1"
alerts <- "a2"
close(alerts)
ctx, cancel := context.WithCancel(context.Background())
cancel()
n := &toggleNotifier{}
var overflow atomic.Int64
deliver(ctx, time.Hour, n, alerts, &overflow)
assert.Equal(t, []string{"a1", "a2"}, n.msgs)
}
// PR #180 round 2: when the first receive after close finds the handoff empty,
// overflow counted by the scan was never moved into the summary.
func TestDeliverCountsOverflowWhenTheHandoffIsClosedAndEmpty(t *testing.T) {
alerts := make(chan string)
close(alerts)
n := &recNotifier{}
var overflow atomic.Int64
overflow.Store(3)
deliver(context.Background(), time.Hour, n, alerts, &overflow)
require.Len(t, n.msgs, 1)
assert.Contains(t, n.msgs[0], "3 further alert(s)")
}
+4 -11
View File
@@ -7,7 +7,6 @@ import (
"path/filepath" "path/filepath"
"strings" "strings"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"github.com/mathiasbq/hyperguild/ingestion/internal/textutil" "github.com/mathiasbq/hyperguild/ingestion/internal/textutil"
) )
@@ -17,11 +16,6 @@ func RebuildIndex(brainDir, date string) error {
if err != nil { if err != nil {
return fmt.Errorf("load inventory: %w", err) return fmt.Errorf("load inventory: %w", err)
} }
bfs, err := brainfs.Open(brainDir)
if err != nil {
return fmt.Errorf("open brain: %w", err)
}
defer func() { _ = bfs.Close() }()
total := len(inv[PageTypeConcept]) + len(inv[PageTypeEntity]) + len(inv[PageTypeSource]) total := len(inv[PageTypeConcept]) + len(inv[PageTypeEntity]) + len(inv[PageTypeSource])
var sb strings.Builder var sb strings.Builder
@@ -40,7 +34,7 @@ func RebuildIndex(brainDir, date string) error {
label := strings.ToUpper(string(pt)[:1]) + string(pt)[1:] label := strings.ToUpper(string(pt)[:1]) + string(pt)[1:]
fmt.Fprintf(&sb, "## %s\n\n", label) fmt.Fprintf(&sb, "## %s\n\n", label)
for _, e := range entries { for _, e := range entries {
summary := pageFirstSentence(bfs, e) summary := pageFirstSentence(brainDir, e)
if summary != "" { if summary != "" {
fmt.Fprintf(&sb, "- [[%s|%s]] — %s\n", e.Slug, e.Title, summary) fmt.Fprintf(&sb, "- [[%s|%s]] — %s\n", e.Slug, e.Title, summary)
} else { } else {
@@ -54,10 +48,9 @@ func RebuildIndex(brainDir, date string) error {
return os.WriteFile(dest, []byte(sb.String()), 0o644) return os.WriteFile(dest, []byte(sb.String()), 0o644)
} }
// pageFirstSentence reads the page through the brain root, regular files func pageFirstSentence(brainDir string, e Entry) string {
// only (hyperguild#157). path := filepath.Join(brainDir, "wiki", string(e.Type), e.Slug+".md")
func pageFirstSentence(bfs *brainfs.Brain, e Entry) string { content, err := os.ReadFile(path)
content, err := bfs.ReadRegular("wiki/" + string(e.Type) + "/" + e.Slug + ".md")
if err != nil { if err != nil {
return "" return ""
} }
+15 -23
View File
@@ -3,45 +3,36 @@ package wiki
import ( import (
"bufio" "bufio"
"bytes"
"errors"
"fmt" "fmt"
"io/fs" "os"
"path/filepath"
"strings" "strings"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
) )
// LoadInventory walks brain/wiki/ and returns all pages grouped by type. // LoadInventory walks brain/wiki/ and returns all pages grouped by type.
// Missing subdirectories are silently skipped. Pages are listed and read // Missing subdirectories are silently skipped.
// through the brain root, regular files only (hyperguild#157); a type
// directory that is a symlink is an error.
func LoadInventory(brainDir string) (map[PageType][]Entry, error) { func LoadInventory(brainDir string) (map[PageType][]Entry, error) {
result := map[PageType][]Entry{ result := map[PageType][]Entry{
PageTypeConcept: {}, PageTypeConcept: {},
PageTypeEntity: {}, PageTypeEntity: {},
PageTypeSource: {}, PageTypeSource: {},
} }
bfs, err := brainfs.Open(brainDir)
if err != nil {
return nil, fmt.Errorf("open brain: %w", err)
}
defer func() { _ = bfs.Close() }()
for pt := range result { for pt := range result {
dir := "wiki/" + string(pt) dir := filepath.Join(brainDir, "wiki", string(pt))
names, err := bfs.ReadDirRegular(dir) entries, err := os.ReadDir(dir)
if errors.Is(err, fs.ErrNotExist) { if os.IsNotExist(err) {
continue continue
} }
if err != nil { if err != nil {
return nil, fmt.Errorf("read dir %s: %w", dir, err) return nil, fmt.Errorf("read dir %s: %w", dir, err)
} }
for _, name := range names { for _, e := range entries {
if !strings.HasSuffix(name, ".md") { if e.IsDir() || !strings.HasSuffix(e.Name(), ".md") {
continue continue
} }
slug := strings.TrimSuffix(name, ".md") slug := strings.TrimSuffix(e.Name(), ".md")
title, aliases := readFrontmatter(bfs, dir+"/"+name, slug) path := filepath.Join(dir, e.Name())
title, aliases := readFrontmatter(path, slug)
result[pt] = append(result[pt], Entry{Slug: slug, Title: title, Aliases: aliases, Type: pt}) result[pt] = append(result[pt], Entry{Slug: slug, Title: title, Aliases: aliases, Type: pt})
} }
} }
@@ -50,14 +41,15 @@ func LoadInventory(brainDir string) (map[PageType][]Entry, error) {
// readFrontmatter extracts title and aliases from YAML frontmatter. // readFrontmatter extracts title and aliases from YAML frontmatter.
// Falls back to slug for title and empty aliases on any error. // Falls back to slug for title and empty aliases on any error.
func readFrontmatter(bfs *brainfs.Brain, rel, fallbackSlug string) (title string, aliases []string) { func readFrontmatter(path, fallbackSlug string) (title string, aliases []string) {
title = fallbackSlug title = fallbackSlug
content, err := bfs.ReadRegular(rel) f, err := os.Open(path)
if err != nil { if err != nil {
return return
} }
defer f.Close() //nolint:errcheck
scanner := bufio.NewScanner(bytes.NewReader(content)) scanner := bufio.NewScanner(f)
inFM := false inFM := false
inAliases := false inAliases := false
for scanner.Scan() { for scanner.Scan() {
-79
View File
@@ -1,79 +0,0 @@
package wiki
import (
"fmt"
"os"
"path/filepath"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs/brainfstest"
)
// hyperguild#157 slice 4: LoadInventory listed wiki/<type>/ with os.ReadDir
// and opened each .md by path, so a symlinked page put the target's title and
// aliases into the inventory (and the LLM prompts built from it), and a FIFO
// hung it. RebuildIndex read first sentences the same way.
func TestLoadInventoryRefusesSymlinksAndSurvivesSwapRaces(t *testing.T) {
brain := t.TempDir()
dir := filepath.Join(brain, "wiki", "concepts")
require.NoError(t, os.MkdirAll(dir, 0o755))
require.NoError(t, os.WriteFile(filepath.Join(dir, "ok.md"), []byte("---\ntitle: zqinside\n---\n"), 0o644))
require.NoError(t, os.Symlink(brainfstest.Outside(t), filepath.Join(dir, "leak.md")))
inBrain := brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md"))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "s.md"), filepath.Join(dir, "inlink.md")))
inv, err := LoadInventory(brain)
require.NoError(t, err)
got := fmt.Sprintf("%+v", inv)
assert.Contains(t, got, "zqinside", "positive control")
assert.NotContains(t, got, brainfstest.Sentinel)
assert.NotContains(t, got, "leak", "a symlinked page is not listed")
assert.NotContains(t, got, "inlink", "a page linked to another brain file is not listed")
p := filepath.Join(dir, "r.md")
read := func() string {
inv, _ := LoadInventory(brain)
return fmt.Sprintf("%+v", inv)
}
brainfstest.AssertNoLeak(t, p, "---\ntitle: zqinside\n---\n", 1500*time.Millisecond, read)
brainfstest.AssertNoLeakFrom(t, p, "---\ntitle: zqinside\n---\n", inBrain, 1500*time.Millisecond, read)
brainfstest.AssertNoHang(t, p, "---\ntitle: zqinside\n---\n", 1500*time.Millisecond, 6*time.Second, func() { _ = read() })
}
func TestLoadInventoryReturnsErrNotRegularForASymlinkedTypeDir(t *testing.T) {
brain := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki"), 0o755))
require.NoError(t, os.MkdirAll(filepath.Join(brain, "raw"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(brain, "raw", "r.md"), []byte("---\ntitle: zqrawonly\n---\n"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("..", "raw"), filepath.Join(brain, "wiki", "concepts")))
inv, err := LoadInventory(brain)
require.ErrorIs(t, err, brainfs.ErrNotRegular)
assert.NotContains(t, fmt.Sprintf("%+v", inv), "zqrawonly")
}
func TestRebuildIndexDoesNotReadAPageSwappedForASymlink(t *testing.T) {
brain := t.TempDir()
dir := filepath.Join(brain, "wiki", "concepts")
require.NoError(t, os.MkdirAll(dir, 0o755))
p := filepath.Join(dir, "r.md")
require.NoError(t, os.WriteFile(p, []byte("---\ntitle: R\n---\nzqinside sentence.\n"), 0o644))
require.NoError(t, RebuildIndex(brain, "2026-10-04"))
idx, err := os.ReadFile(filepath.Join(brain, "wiki", "index.md"))
require.NoError(t, err)
require.Contains(t, string(idx), "zqinside", "positive control: the summary is read")
read := func() string {
_ = RebuildIndex(brain, "2026-10-04")
b, _ := os.ReadFile(filepath.Join(brain, "wiki", "index.md"))
return string(b)
}
brainfstest.AssertNoLeak(t, p, "---\ntitle: R\n---\nzqinside sentence.\n", 1500*time.Millisecond, read)
inBrain := brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md"))
brainfstest.AssertNoLeakFrom(t, p, "---\ntitle: R\n---\nzqinside sentence.\n", inBrain, 1500*time.Millisecond, read)
brainfstest.AssertNoHang(t, p, "---\ntitle: R\n---\nzqinside sentence.\n", 1500*time.Millisecond, 6*time.Second, func() { _ = read() })
}