github · nocturne_memoryd4f82fcddfc1

Evidence

if __name__ == "__main__":

    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8233)

Remediation

Bind to 127.0.0.1 for local-only access. If cross-host access is truly required, put the service behind an authenticated reverse proxy rather than exposing it on 0.0.0.0.

Evidence

port = int(os.environ.get("WEB_PORT", "8233"))

            config = uvicorn.Config(

                build_web_app(), host="0.0.0.0", port=port, log_level="warning",

            )

            web_server = uvicorn.Server(config)

Remediation

Bind to 127.0.0.1 for local-only access. If cross-host access is truly required, put the service behind an authenticated reverse proxy rather than exposing it on 0.0.0.0.

Evidence

backend-api:

    build:

      context: ./backend

    command: uvicorn main:app --host 0.0.0.0 --port 8233

    environment:

      DATABASE_URL: postgresql+asyncpg://${POSTGRES_USER}:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB}?ssl=disable

      API_TOKEN: ${API_TOKEN:-}

Remediation

Bind to 127.0.0.1 for local-only access. If cross-host access is truly required, put the service behind an authenticated reverse proxy rather than exposing it on 0.0.0.0.

Evidence

return content

    except Exception as e:

        # Catch both ValueError (not found) and other exceptions

        return f"Error: {str(e)}"

@write_tool()

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

return "\n".join(lines)

    except Exception as e:

        return f"Error generating glossary index: {str(e)}"

async def generate_diagnostic_view(domain: str, days_stale: int = 30, max_children: int = 10) -> str:

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

return "\n".join(lines)

    except Exception as e:

        return f"Error generating recent memories view: {str(e)}"

async def generate_glossary_index_view() -> str:

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

except ValueError as e:

        return f"Error: {str(e)}"

    except Exception as e:

        return f"Error: {str(e)}"

@mcp.tool()

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

except ValueError as e:

        return f"Error: {str(e)}"

    except Exception as e:

        return f"Error: {str(e)}"

@write_tool()

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

except ValueError as e:

        return f"Error: {str(e)}"

    except Exception as e:

        return f"Error: {str(e)}"

@write_tool()

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

except ValueError as e:

        return f"Error: {str(e)}"

    except Exception as e:

        return f"Error: {str(e)}"

@write_tool()

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

except ValueError as e:

        return f"Error: {str(e)}"

    except Exception as e:

        return f"Error: {str(e)}"

@write_tool()

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

)

    except ValueError as e:

        return f"Error: {str(e)}"

    except Exception as e:

        return f"Error: {str(e)}"

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

return "\n".join(lines).strip()

    except Exception as e:

        return f"Error generating diagnostic view: {str(e)}"

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

return msg

    except ValueError as e:

        return f"Error: {str(e)}"

    except Exception as e:

        return f"Error: {str(e)}"

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

return "\n".join(lines)

    except Exception as e:

        return f"Error generating index: {str(e)}"

async def generate_recent_memories_view(limit: int = 10) -> str:

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

return "\n".join(lines)

    except ValueError as e:

        return f"Error: {str(e)}"

    except Exception as e:

        return f"Error: {str(e)}"

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

return message

    except ValueError as e:

        return f"Error: {str(e)}"

    except Exception as e:

        return f"Error: {str(e)}"

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

return msg

    except ValueError as e:

        return f"Error: {str(e)}"

    except Exception as e:

        return f"Error: {str(e)}"

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

return "\n".join(lines)

    except Exception as e:

        return f"Error: {str(e)}"

# =============================================================================

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

@router.put("/node")

async def update_node(

    path: str = Query(...),

    domain: str = Query("core"),

    body: NodeUpdate = ...

):

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

if scope["type"] != "http":

                await self.backend(scope, receive, send)

                return

            path: str = scope.get("path", "/")

            if any(path == p or path.startswith(p + "/") for p in backend_prefixes):

                await self.backend(scope, receive, send)

                return

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

@router.get("/node")

async def get_node(

    path: str = Query("", description="URI path like 'nocturne' or 'nocturne/salem'"),

    domain: str = Query("core"),

    nav_only: bool = Query(False, description="Skip expensive processing if only navigating tree")

):

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

# Stage 1: 构建 React 应用

FROM node:20-alpine AS build

WORKDIR /app

# 先复制依赖声明，利用 Docker 缓存层

COPY package*.json ./

RUN npm ci

# 复制源码和构建配置

COPY index.html ./

COPY vite.config.js ./

COPY tailwind.config.js ./

COPY postcss.config.js ./

COPY src/ ./src/

RUN npm run build

# Stage 2: 生产环境 nginx 服务

FROM nginx:alpine AS production

# 用自定义配置覆盖默认配置

RUN rm /etc/nginx/conf.d/default.conf

COPY nginx.conf /etc/nginx/conf.d/default.conf

# 从构建阶段复制产物

COPY --from=build /app/dist /usr/share/nginx/html

EXPOSE 80

Remediation

Create and switch to a non-root user before the CMD / ENTRYPOINT: RUN adduser --system --uid 1000 app USER 1000 Or reuse the base image's shipped non-root account (e.g. `USER nobody`, `USER nonroot` on distroless). Multi-stage builds only need the USER directive in the final stage.

Evidence

steps:

      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5

        with:

          python-version: "3.12"

          cache: pip

Remediation

Pin every `uses:` to a 40-character commit SHA. Trailing comment with the version helps reviewers: `uses: actions/checkout@8e5e7e5ab8b370d6c329ec480221332ada57f0ab # v4.1.1` Automate the migration with `pinact` (https://github.com/suzuki-shunsuke/pinact) or `ratchet` (https://github.com/sethvargo/ratchet). Add a `pinact run --check` pre-commit hook so future PRs stay pinned. Re-pin when the action releases a new version — Dependabot can do this automatically with `version-update-strategy: inc

Evidence

steps:

      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5

        with:

          python-version: ${{ matrix.python-version }}

          cache: pip

Remediation

Pin every `uses:` to a 40-character commit SHA. Trailing comment with the version helps reviewers: `uses: actions/checkout@8e5e7e5ab8b370d6c329ec480221332ada57f0ab # v4.1.1` Automate the migration with `pinact` (https://github.com/suzuki-shunsuke/pinact) or `ratchet` (https://github.com/sethvargo/ratchet). Add a `pinact run --check` pre-commit hook so future PRs stay pinned. Re-pin when the action releases a new version — Dependabot can do this automatically with `version-update-strategy: inc

Evidence

--health-retries 5

    steps:

      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5

        with:

Remediation

Pin every `uses:` to a 40-character commit SHA. Trailing comment with the version helps reviewers: `uses: actions/checkout@8e5e7e5ab8b370d6c329ec480221332ada57f0ab # v4.1.1` Automate the migration with `pinact` (https://github.com/suzuki-shunsuke/pinact) or `ratchet` (https://github.com/sethvargo/ratchet). Add a `pinact run --check` pre-commit hook so future PRs stay pinned. Re-pin when the action releases a new version — Dependabot can do this automatically with `version-update-strategy: inc

Evidence

python-version: ["3.10", "3.12"]

    steps:

      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5

        with:

Remediation

Pin every `uses:` to a 40-character commit SHA. Trailing comment with the version helps reviewers: `uses: actions/checkout@8e5e7e5ab8b370d6c329ec480221332ada57f0ab # v4.1.1` Automate the migration with `pinact` (https://github.com/suzuki-shunsuke/pinact) or `ratchet` (https://github.com/sethvargo/ratchet). Add a `pinact run --check` pre-commit hook so future PRs stay pinned. Re-pin when the action releases a new version — Dependabot can do this automatically with `version-update-strategy: inc

Evidence

# ── SSE follow-up message ───────────────────────────────────────

            # Restore namespace from the session established at connect time.

            session_id = request.query_params.get("session_id", "")

            ns = _sse_sessions.get(session_id, "")

            # Fall back to per-request header/param if no session entry.

            if not ns:

                ns = request.headers.get("x-namespace", "")

Remediation

Bind session IDs to user IDs in every key: # BAD ddb.get_item(Key={"session_id": session_id}) data = sessions[session_id] redis.get(session_id) # GOOD ddb.get_item(Key={"user_id": user_id, "session_id": session_id}) data = sessions[f"{user_id}:{session_id}"] redis.get(f"{user_id}:{session_id}") The user_id should come from the verified authorization token (JWT claim, validated session cookie) — never from a request parameter the user can choose.

Evidence

docs_url="/docs",

        openapi_url="/openapi.json",

    )

    api.add_middleware(

        CORSMiddleware,

        allow_origins=["*"],

        allow_credentials=True,

        allow_methods=["*"],

        allow_headers=["*"],

    )

Remediation

Replace wildcard / reflected origin with an explicit allowlist of trusted origins: cors({ origin: ["https://app.example.com", "https://admin.example.com"], credentials: true, }) Or, if the API genuinely should be accessible from any origin, drop credentials and use bearer tokens in the `Authorization` header instead — that combo is browser-safe.

Evidence

pass

        finally:

            try:

                process.stdin.close()

            except Exception:

                pass

    def forward_stdout():

        """Forward stdout from the subprocess, normalizing CRLF to LF."""

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

try:

                    parsed = urlparse(db_url)

                    if parsed.password:

                        db_url = db_url.replace(f":{parsed.password}@", ":***@")

                except Exception:

                    pass

            raise RuntimeError(

                f"Failed to connect to database.\n"

                f"  URL: {db_url}\n"

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

output_parts.append("")

        output_parts.append("---")

        output_parts.append("")

        output_parts.append(recent_view)

    except Exception:

        pass

    return "\n".join(output_parts)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

path = self._path(session_id)

            if os.path.exists(path):

                try:

                    os.remove(path)

                except OSError:

                    pass

        return val

# Populated when the FastMCP "endpoint" SSE event is intercepted on /sse.

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

def _remove_changeset(self):

        p = self._changeset_path

        try:

            os.remove(p)

        except FileNotFoundError:

            pass

        except PermissionError:

            os.chmod(p, stat.S_IWRITE)

            os.remove(p)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

for (name,) in result.all():

            try:

                num = int(name)

                max_num = max(max_num, num)

            except ValueError:

                pass

        return max_num + 1

    async def _would_create_cycle(

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

lines.append("GLOSSARY (keywords detected in this content)")

                lines.append("")

                lines.extend(lines_to_add)

                lines.append("")

    except Exception:

        pass

    if children:

        lines.append("=" * 60)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if not data:

                    break

                process.stdin.write(data)

                process.stdin.flush()

        except Exception:

            pass

        finally:

            try:

                process.stdin.close()

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

def __setitem__(self, session_id: str, namespace: str) -> None:

        try:

            with open(self._path(session_id), "w", encoding="utf-8") as f:

                f.write(namespace)

        except OSError:

            pass

    def get(self, session_id: str, default: str = "") -> str:

        if not session_id:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if os.path.exists(path):

            try:

                with open(path, "r", encoding="utf-8") as f:

                    return f.read()

            except OSError:

                pass

        return default

    def pop(self, session_id: str, default: str = "") -> str:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

# Skip \r characters (convert CRLF to LF)

                if data != b'\r':

                    sys.stdout.buffer.write(data)

                    sys.stdout.buffer.flush()

        except Exception:

            pass

    # Start threads for bidirectional communication

    stdin_thread = threading.Thread(target=forward_stdin, daemon=True)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

f = (self.dist / path.lstrip("/")).resolve()

                if path != "/" and f.is_file() and f.is_relative_to(self.dist):

                    await FileResponse(f)(scope, receive, send)

                    return

            except (ValueError, OSError):

                pass

            index_file = self.dist / "index.html"

            if index_file.is_file():

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.