mirror of
https://github.com/Heretek-AI/cod.ai.git
synced 2026-11-07 10:46:11 -10:00
Validate session ids before building a session path
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_015NGPpYxuPWkyJUoefut3Wo
This commit is contained in:
3 files changed
+16
-7
No files matched your search
+11
-1
@@ -11,11 +11,14 @@ and thinking blocks stay valid.
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import secrets
|
||||
from typing import Any
|
||||
|
||||
SESSIONS_DIR = Path(".cod") / "sessions"
|
||||
SESSION_ID_PATTERN = re.compile(r"[0-9]{8}-[0-9]{6}-[0-9a-f]{6}")
|
||||
|
||||
|
||||
class SessionError(RuntimeError):
|
||||
@@ -89,7 +92,14 @@ class Session:
|
||||
"""Loads a session by id, or the most recent one when no id is given."""
|
||||
directory = Path(root) / SESSIONS_DIR
|
||||
if session_id:
|
||||
return cls.load(directory / f"{session_id}.jsonl")
|
||||
# Ids come from the command line or a URL, so only the generated form is accepted.
|
||||
if not SESSION_ID_PATTERN.fullmatch(session_id):
|
||||
raise SessionError(f"invalid session id {session_id!r}")
|
||||
base = os.path.realpath(directory)
|
||||
path = os.path.realpath(os.path.join(base, f"{session_id}.jsonl"))
|
||||
if not path.startswith(base + os.sep):
|
||||
raise SessionError(f"invalid session id {session_id!r}")
|
||||
return cls.load(path)
|
||||
candidates = sorted(directory.glob("*.jsonl")) if directory.is_dir() else []
|
||||
if not candidates:
|
||||
raise SessionError(f"no sessions in {directory}")
|
||||
|
||||
@@ -15,7 +15,6 @@ import asyncio
|
||||
from dataclasses import dataclass, field
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
import secrets
|
||||
from typing import Any, AsyncIterator, Literal
|
||||
|
||||
@@ -47,7 +46,6 @@ from src.agent.tools import ToolError, Toolbox
|
||||
MAX_PROMPT_CHARS = 1_000_000
|
||||
# Tool output shown in the browser; the model still gets the full result.
|
||||
MAX_OUTPUT_PREVIEW = 4_000
|
||||
SESSION_ID = re.compile(r"^[0-9]{8}-[0-9]{6}-[0-9a-f]{6}$")
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -158,8 +156,6 @@ class AgentService:
|
||||
def get(self, session_id: str) -> LiveSession:
|
||||
if session_id in self.live:
|
||||
return self.live[session_id]
|
||||
if not SESSION_ID.match(session_id):
|
||||
raise HTTPException(404, "no such session")
|
||||
try:
|
||||
session = Session.find(self.settings.root, session_id)
|
||||
except SessionError:
|
||||
|
||||
+5
-2
@@ -281,8 +281,11 @@ class SessionTests(TempProject):
|
||||
def test_missing_and_corrupt_sessions(self):
|
||||
with self.assertRaises(SessionError):
|
||||
Session.find(self.root)
|
||||
for bad_id in ("nope", "../../etc/passwd", "20260101-000000-abcdef/../x"):
|
||||
with self.assertRaises(SessionError):
|
||||
Session.find(self.root, bad_id)
|
||||
with self.assertRaises(SessionError):
|
||||
Session.find(self.root, "nope")
|
||||
Session.find(self.root, "20260101-000000-abcdef")
|
||||
bad = self.write(".cod/sessions/bad.jsonl", "{broken\n")
|
||||
with self.assertRaises(SessionError):
|
||||
Session.load(bad)
|
||||
@@ -623,7 +626,7 @@ class CliTests(TempProject):
|
||||
self.assertIn("authentication", err)
|
||||
code, _, err = self.cod("--resume", "missing", "-p", "x", client=FakeClient(refusal))
|
||||
self.assertEqual(code, 1)
|
||||
self.assertIn("no session file", err)
|
||||
self.assertIn("invalid session id", err)
|
||||
|
||||
def test_serve_refuses_public_hosts(self):
|
||||
self.assertTrue(cli.is_loopback_host("::1"))
|
||||
|
||||
Reference in new issue
Block a user