github · vault-cortex4bd51c642a60

Evidence

const started = flags.yes

    ? false

    : await offerComposeUp({ targetDir, port }, deps)

  prompts.print(

    buildLocalConnectMessage({ targetDir, token, started, port, tokenWritten }),

  )

  return 0

}

Remediation

Do not return raw secret material from MCP tools or emit it to logs. Redact via a helper (`mask`, `redact`, `sanitize`), hash via `hashlib` / `crypto.createHash`, or expose only the length / a short prefix. For Secrets Manager and KMS responses, avoid returning `SecretString` / `Plaintext` from tool handlers — return an opaque reference or a derived value instead.

Evidence

obsidianAuthToken === ""

      ? false

      : await offerComposeUp({ targetDir, port }, deps)

  prompts.print(

    buildRemoteConnectMessage({

      targetDir,

      token,

      publicUrl,

      started,

      obsidianTokenMissing: obsidianAuthToken === "",

      tokenWritten,

    }),

  )

  return 0

Remediation

Do not return raw secret material from MCP tools or emit it to logs. Redact via a helper (`mask`, `redact`, `sanitize`), hash via `hashlib` / `crypto.createHash`, or expose only the length / a short prefix. For Secrets Manager and KMS responses, avoid returning `SecretString` / `Plaintext` from tool handlers — return an opaque reference or a derived value instead.

Evidence

/** MCP prompt definitions — user-initiated guided workflows over the vault.

 *

 * Prompts are the counterpart to tools: tools are model-driven, while prompts

 * are user-initiated — the client surfaces them as slash commands, a "+" menu,

 * or similar — and assemble live vault content at invocation time. Each handler

 * gathers from the same data layer the tools use, then returns a single text

 * message — no embedded procedure that can drift, just live content plus thin,

 * durable instruction

Remediation

Wrap the parameter in `<untrusted>...</untrusted>` so the LLM treats it as data, run it through `escape_for_prompt` / `sanitize_prompt`, or pass via `json.dumps` / `JSON.stringify`. If you control the prompt template, prefer keeping it static and putting user data in a separate `role: "user"` message.

Evidence

/** MCP session routes — transport lifecycle, session creation, request routing. */

import { Router } from "express"

import type { Request, Response } from "express"

import { randomUUID } from "node:crypto"

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"

import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"

import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"

import { requireBearerAuth } from "@modelcontextprot

Remediation

Apply auth middleware at the route or router level: - Express / Fastify / Koa: `passport.authenticate(...)`, `requireAuth`, `verifyToken`, or an equivalent JWT middleware applied via `router.use(authMw)` or as a per-route handler. - FastAPI: `Depends(get_current_user)`, `OAuth2PasswordBearer`, `HTTPBearer`, or `verify_jwt` dependency. - Flask: `@login_required`, `@auth_required`, `@jwt_required`, or call `verify_jwt_in_request()` in the handler. Mounting MCP behind a s

Remediation

Typosquat: verify you meant the popular package. If so, correct the spelling; if you truly intended the less-common name, suppress with an inline waiver. Stale release: check whether the package has a maintained fork or successor. If no patched release exists, vendor the code or migrate to an active alternative before the unmaintained code accrues unfixed CVEs.

Remediation

Typosquat: verify you meant the popular package. If so, correct the spelling; if you truly intended the less-common name, suppress with an inline waiver. Stale release: check whether the package has a maintained fork or successor. If no patched release exists, vendor the code or migrate to an active alternative before the unmaintained code accrues unfixed CVEs.

Evidence

"CMD",

          "node",

          "-e",

          "fetch('http://127.0.0.1:8000/healthz').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))",

        ]

      interval: 15s

      timeout: 5s

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

const harness = await setupHarness()

    vi.mocked(isInitializeRequest).mockReturnValue(false)

    const response = await fetch(harness.url(), {

      method: "POST",

      headers: baseHeaders,

      body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/list" }),

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

sessionId,

    })

    // The session should be gone — confirm with a GET that should 404.

    const followUp = await fetch(harness.url(), {

      method: "GET",

      headers: { ...baseHeaders, "mcp-session-id": sessionId },

    })

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

const customConfig = loadConfig({ MEMORY_DIR: "Profile" })

      const harness = await setupHarness({ config: customConfig })

      vi.mocked(isInitializeRequest).mockReturnValue(true)

      await fetch(harness.url(), {

        method: "POST",

        headers: { ...baseHeaders },

        body: JSON.stringify(initializeBody),

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

const harness = await setupHarness()

    const { sessionId, transport } = await createSession(harness)

    const response = await fetch(harness.url(), {

      method: "DELETE",

      headers: { ...baseHeaders, "mcp-session-id": sessionId },

    })

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

"CMD",

          "node",

          "-e",

          "fetch('http://127.0.0.1:8000/healthz').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))",

        ]

      interval: 15s

      timeout: 5s

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

"CMD",

          "node",

          "-e",

          "fetch('http://127.0.0.1:8000/healthz').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))",

        ]

      interval: 15s

      timeout: 5s

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

it("returns 404 when the mcp-session-id header is missing", async () => {

    const harness = await setupHarness()

    const response = await fetch(harness.url(), {

      method: "DELETE",

      headers: baseHeaders,

    })

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

"CMD",

          "node",

          "-e",

          "fetch('http://127.0.0.1:8000/healthz').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))",

        ]

      interval: 15s

      timeout: 5s

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

"CMD",

          "node",

          "-e",

          "fetch('http://127.0.0.1:8000/healthz').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))",

        ]

      interval: 15s

      timeout: 5s

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

it("returns 401 when bearer auth rejects", async () => {

    const harness = await setupHarness({ authMiddleware: denyAuth })

    const response = await fetch(harness.url(), {

      method: "DELETE",

      headers: { ...baseHeaders, "mcp-session-id": "anything" },

    })

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

it("returns 401 when bearer auth rejects", async () => {

    const harness = await setupHarness({ authMiddleware: denyAuth })

    const response = await fetch(harness.url(), {

      method: "GET",

      headers: { ...baseHeaders, "mcp-session-id": "anything" },

    })

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

it("returns 404 when the session id is unknown", async () => {

    const harness = await setupHarness()

    const response = await fetch(harness.url(), {

      method: "GET",

      headers: { ...baseHeaders, "mcp-session-id": "missing" },

    })

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

const createSession = async (

  harness: Harness,

): Promise<{ sessionId: string; transport: TransportMock }> => {

  const response = await fetch(harness.url(), {

    method: "POST",

    headers: baseHeaders,

    body: JSON.stringify(initializeBody),

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

const { sessionId, transport } = await createSession(harness)

    transport.handleRequest.mockClear()

    const response = await fetch(harness.url(), {

      method: "GET",

      headers: { ...baseHeaders, "mcp-session-id": sessionId },

    })

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

it("returns 404 when the session id is unknown", async () => {

    const harness = await setupHarness()

    const response = await fetch(harness.url(), {

      method: "DELETE",

      headers: { ...baseHeaders, "mcp-session-id": "ghost" },

    })

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

it("returns 404 when the mcp-session-id header is missing", async () => {

    const harness = await setupHarness()

    const response = await fetch(harness.url(), {

      method: "GET",

      headers: baseHeaders,

    })

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

// The session should be gone from the map — verified via a follow-up

    // GET that should now 404.

    const followUp = await fetch(harness.url(), {

      method: "GET",

      headers: { ...baseHeaders, "mcp-session-id": sessionId },

    })

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

it("returns 401 and never enters the route handler when bearer auth rejects", async () => {

    const harness = await setupHarness({ authMiddleware: denyAuth })

    const response = await fetch(harness.url(), {

      method: "POST",

      headers: baseHeaders,

      body: JSON.stringify(initializeBody),

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

vi.mocked(isInitializeRequest).mockReturnValue(false)

    const followUp = { jsonrpc: "2.0", id: 2, method: "tools/list" }

    const response = await fetch(harness.url(), {

      method: "POST",

      headers: { ...baseHeaders, "mcp-session-id": sessionId },

      body: JSON.stringify(followUp),

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

it("returns 404 when the session id is unknown", async () => {

    const harness = await setupHarness()

    const response = await fetch(harness.url(), {

      method: "POST",

      headers: { ...baseHeaders, "mcp-session-id": "ghost-session" },

      body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/list" }),

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

Returns: Confirmation message.`,

      inputSchema: {

        path: z.string().min(1).describe("Vault-relative path for the note"),

        body: z

          .string()

          .describe("Markdown body content (no frontmatter fences)"),

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

Returns: Confirmation message.`,

      inputSchema: {

        path: z.string().min(1).describe("Vault-relative path to the note"),

        properties: z

          .record(z.string().min(1), z.unknown())

          .describe(

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

Returns: Confirmation message with the number of lines removed and a truncated preview of the deleted text.`,

      inputSchema: {

        path: z.string().min(1).describe("Vault-relative path to the note"),

        start_anchor: z

          .string()

          .min(1)

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

Returns: JSON with path (the queried note), outgoing_links (array of { path, title, exists, kind, bytes }, sorted by target path), and count. kind is "note" or "asset". bytes is the on-disk file size (null for broken links and assets).`,

      inputSchema: {

        path: z.string().min(1).describe("Vault-relative path to the note"),

      },

      annotations: {

        readOnlyHint: true,

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

Returns: Confirmation message.`,

      inputSchema: {

        path: z.string().min(1).describe("Vault-relative path to the note"),

        operation: z

          .enum(["append", "prepend", "replace", "insert_before"])

          .describe("Patch operation to apply"),

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

Returns: Confirmation message with replacement count.`,

      inputSchema: {

        path: z.string().min(1).describe("Vault-relative path to the note"),

        old_text: z

          .string()

          .min(1)

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

"lint:fix": "eslint --fix .",

    "test": "vitest run",

    "test:watch": "vitest",

    "prepare": "husky"

  },

  "lint-staged": {

    "*.{json,md}": "prettier --write",

Remediation

Prefer libraries that do not require install-time code execution: - Drop `postinstall`/`preinstall`/`prepare` scripts if the work can happen at runtime or build-time instead. - Ship pre-built native binaries rather than compiling via a custom `cmdclass` or `build_ext` override. - For Dockerfiles: replace `RUN curl … | sh` with a pinned download + checksum verification + explicit `RUN` of a named script. - If the hook is unavoidable, document exactly what it does so downstream reviewers

Evidence

# Local Quickstart

Run Vault Cortex on your machine against a local Obsidian vault. No cloud, no

Obsidian Sync — just Docker and a folder of `.md` files.

> **Fastest path:** `npx vault-cortex@latest init` does all of the below

> interactively — generates the token and config files, starts the server, and

> prints the connection details. The steps below are the manual equivalent.

## Prerequisites

- [Docker](https://docs.docker.com/get-docker/) (v20.10+)

- An Obsidian vault (or any folder of M

Remediation

Declare a real authentication mechanism in the manifest, matching what the running server actually enforces: - `"auth": "bearer"` with a token scheme documented for callers - `"auth": "oauth"` / `"oauth2": { ... }` for delegated flows - `"apiKey": { "header": "X-API-Key", "prefix": "..." }` - `"mtls": true` when client certificates are required If the server is intentionally unauthenticated (stdio-only, local developer tool, trusted-host network), document the assumption in the manifest via a `"

Evidence

<p align="center">

  <img src="./assets/banner.svg" width="720" alt="Vault Cortex">

</p>

<div align="center">

[![CI](https://img.shields.io/github/actions/workflow/status/aliasunder/vault-cortex/ci.yml?branch=main&logo=github&label=CI&cacheSeconds=43200)](https://github.com/aliasunder/vault-cortex/actions/workflows/ci.yml)

[![Gitleaks](https://img.shields.io/github/actions/workflow/status/aliasunder/vault-cortex/gitleaks.yml?branch=main&logo=github&label=Gitleaks&cacheSeconds=43200)](https://g

Remediation

Declare a real authentication mechanism in the manifest, matching what the running server actually enforces: - `"auth": "bearer"` with a token scheme documented for callers - `"auth": "oauth"` / `"oauth2": { ... }` for delegated flows - `"apiKey": { "header": "X-API-Key", "prefix": "..." }` - `"mtls": true` when client certificates are required If the server is intentionally unauthenticated (stdio-only, local developer tool, trusted-host network), document the assumption in the manifest via a `"

Evidence

/** MCP session routes — transport lifecycle, session creation, request routing. */

import { Router } from "express"

import type { Request, Response } from "express"

import { randomUUID } from "node:crypto"

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"

import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"

import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"

import { requireBearerAuth } from "@modelcontextprot

Remediation

Apply CSRF middleware at the route or router level: - Express: `app.use(csurf())` / `csrf-csrf` package - FastAPI: `fastapi-csrf-protect` with `Depends(...)` - Flask: `CSRFProtect(app)` from `flask_wtf.csrf` Or move to bearer-token auth and set `SameSite=Strict` / `SameSite=Lax` on any session cookies. Document the choice in the project README so reviewers can confirm intent.

Evidence

/** OAuth HTTP routes — SDK auth router + consent form handler. */

import express, { Router } from "express"

import type { Request, Response } from "express"

import { mcpAuthRouter } from "@modelcontextprotocol/sdk/server/auth/router.js"

import { safeEqual } from "../../auth.js"

import { renderConsentPage } from "./consent-page.js"

import type { OAuthProvider } from "./oauth-provider.js"

export type OAuthRoutesOptions = {

  authToken: string

  serverUrl: URL

  oauthProvider: OAuthProvider

  se

Remediation

Apply CSRF middleware at the route or router level: - Express: `app.use(csurf())` / `csrf-csrf` package - FastAPI: `fastapi-csrf-protect` with `Depends(...)` - Flask: `CSRFProtect(app)` from `flask_wtf.csrf` Or move to bearer-token auth and set `SameSite=Strict` / `SameSite=Lax` on any session cookies. Document the choice in the project README so reviewers can confirm intent.

Evidence

import {

  chmodSync,

  existsSync,

  mkdirSync,

  readFileSync,

  writeFileSync,

} from "node:fs"

import { join } from "node:path"

import { fileURLToPath } from "node:url"

export type Mode = "local" | "remote"

export type FileToWrite = {

  /** Filename inside the target directory (e.g. "docker-compose.yml"). */

  name: string

  content: string

  /** Unix permission bits for files holding secrets (e.g. 0o600 for .env). */

  mode?: number

}

export type FileWriteResult = {

  name: string

  stat

Remediation

Replace check-then-use with action-then-handle: Python: `try: with open(p) as f: ... except FileNotFoundError: ...` Node: `try { fs.readFileSync(p); } catch (e) { if (e.code === "ENOENT") ... }` For atomic file creation: Python: `os.open(p, os.O_RDWR | os.O_CREAT | os.O_EXCL)` Node: `fs.open(p, "wx")` — fails if file exists, no race. When you genuinely must check first, use `flock` (Python `fcntl`) or a similar per-process advisory lock to make the window uninteresting to a

Evidence

/// <reference path="./.sst/platform/config.d.ts" />

// SST 4 forbids top-level imports in sst.config.ts — everything has

// to be dynamically imported inside `run()`. See readSshPublicKey

// below for the only filesystem access we need.

// Module-level so app() and run() share the same value.

// env-var can't be used here (SST forbids imports outside run()).

const awsRegion = process.env.AWS_REGION ?? "us-east-1"

export default $config({

  app() {

    return {

      name: "vault-cortex",

Remediation

Replace check-then-use with action-then-handle: Python: `try: with open(p) as f: ... except FileNotFoundError: ...` Node: `try { fs.readFileSync(p); } catch (e) { if (e.code === "ENOENT") ... }` For atomic file creation: Python: `os.open(p, os.O_RDWR | os.O_CREAT | os.O_EXCL)` Node: `fs.open(p, "wx")` — fails if file exists, no race. When you genuinely must check first, use `flock` (Python `fcntl`) or a similar per-process advisory lock to make the window uninteresting to a

Evidence

#!/usr/bin/env tsx

/**

 * Deployment helper.

 *

 * Subcommands:

 *   docker:build    Build the vault-mcp image locally

 *   docker:push     Push to GHCR

 *   docker:publish  Build + push

 *   lightsail:up    SCP docker-compose.yml + ~/.config/vault-cortex/.env

 *                   to the VM, then `docker compose pull && up -d` over SSH

 *

 * The image is always `ghcr.io/${GHCR_USER}/vault-mcp:latest`. The

 * Lightsail IP is fetched from AWS (`aws lightsail get-static-ip`)

 * using the stage in `

Remediation

Replace check-then-use with action-then-handle: Python: `try: with open(p) as f: ... except FileNotFoundError: ...` Node: `try { fs.readFileSync(p); } catch (e) { if (e.code === "ENOENT") ... }` For atomic file creation: Python: `os.open(p, os.O_RDWR | os.O_CREAT | os.O_EXCL)` Node: `fs.open(p, "wx")` — fails if file exists, no race. When you genuinely must check first, use `flock` (Python `fcntl`) or a similar per-process advisory lock to make the window uninteresting to a

Evidence

#!/usr/bin/env bash

# Update CHANGELOG.md with release notes for a given version.

# Usage: update-changelog.sh <version> <notes-file>

# Replaces [Unreleased] section if present, otherwise inserts after header.

set -euo pipefail

VERSION="${1:?Usage: update-changelog.sh <version> <notes-file>}"

NOTES_FILE="${2:?Usage: update-changelog.sh <version> <notes-file>}"

NOTES=$(cat "$NOTES_FILE")

DATE=$(date +%Y-%m-%d)

python3 - "$VERSION" "$DATE" "$NOTES" << 'PYTHON'

import sys, re, os

version, date

Remediation

Replace check-then-use with action-then-handle: Python: `try: with open(p) as f: ... except FileNotFoundError: ...` Node: `try { fs.readFileSync(p); } catch (e) { if (e.code === "ENOENT") ... }` For atomic file creation: Python: `os.open(p, os.O_RDWR | os.O_CREAT | os.O_EXCL)` Node: `fs.open(p, "wx")` — fails if file exists, no race. When you genuinely must check first, use `flock` (Python `fcntl`) or a similar per-process advisory lock to make the window uninteresting to a

Evidence

/** MCP session routes — transport lifecycle, session creation, request routing. */

import { Router } from "express"

import type { Request, Response } from "express"

import { randomUUID } from "node:crypto"

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"

import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"

import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"

import { requireBearerAuth } from "@modelcontextprot

Remediation

Add a structured audit-event emit immediately after every destructive sink. Minimum schema: actor, action, target, outcome, request_id. Python: from acme.audit import audit_log @mcp.tool() def delete_record(token: str, record_id: str) -> dict: actor = verify_token(token) db.execute("DELETE FROM records WHERE id = %s", (record_id,)) audit_log( actor=actor.sub, action="delete_record", target=record_id, outcome=

Evidence

/**

 * OAuth 2.1 provider for vault-cortex.

 *

 * - Dynamic client registration (Claude Desktop, Perplexity self-register)

 * - Authorization code flow with PKCE

 * - JWT access tokens (HS256, verifiable by Lambda + Express)

 * - Backward-compatible static token verification (MCP_AUTH_TOKEN)

 * - SQLite persistence for refresh tokens + clients (survives restarts)

 * - Consent page gated by the server's auth token

 */

import Database from "better-sqlite3"

import { randomUUID, randomBytes } from 

Remediation

Add a structured audit-event emit immediately after every destructive sink. Minimum schema: actor, action, target, outcome, request_id. Python: from acme.audit import audit_log @mcp.tool() def delete_record(token: str, record_id: str) -> dict: actor = verify_token(token) db.execute("DELETE FROM records WHERE id = %s", (record_id,)) audit_log( actor=actor.sub, action="delete_record", target=record_id, outcome=

Evidence

import {

  describe,

  it,

  expect,

  beforeEach,

  afterEach,

  onTestFinished,

  vi,

} from "vitest"

import express from "express"

import { randomUUID } from "node:crypto"

import type { Server } from "node:http"

import type { AddressInfo } from "node:net"

import { createMcpRouter } from "../mcp-router.js"

import { loadConfig } from "../../config.js"

import type { SearchIndex } from "../../search/search-index.js"

import type { OAuthServerProvider } from "@modelcontextprotocol/sdk/server/auth/p

Remediation

Add a structured audit-event emit immediately after every destructive sink. Minimum schema: actor, action, target, outcome, request_id. Python: from acme.audit import audit_log @mcp.tool() def delete_record(token: str, record_id: str) -> dict: actor = verify_token(token) db.execute("DELETE FROM records WHERE id = %s", (record_id,)) audit_log( actor=actor.sub, action="delete_record", target=record_id, outcome=