Skip to content

C
coderai
Commit f97459fc authored by Stefy Lanza (nextime / spora )'s avatar Stefy Lanza (nextime / spora )
Browse files
Options

admin: sanitize model-upload paths + atomic auth.json read-modify-write 

The model-upload endpoint joined a client-supplied filename straight onto
the cache dir, so an admin-authenticated request with a traversal filename
(or upload_id) could write outside it. Reduce both to a safe basename,
reject separators/.., and add a commonpath containment check before
committing the upload.

SessionManager only locked the write half of each load->mutate->save, so
concurrent writers could clobber each other's changes (lost sessions or
tokens). Add update_auth_data(mutator), which holds the lock across the
whole read-modify-write and persists only when the mutator asks to; route
every mutating method (and the token create/delete endpoints) through it.
Read-only callers keep the lock-free load since writes are atomic via
os.replace. While migrating the token endpoints, switch IDs to max+1 (no
reuse after deletion) and to timezone-aware timestamps.
Co-Authored-By: 's avatarClaude Opus 4.8 <noreply@anthropic.com>
parent 4754beff
Hide whitespace changes
Inline Side-by-side
Showing with 183 additions and 161 deletions
codai/admin/auth.py
View file @ f97459fc
......@@ -142,22 +142,47 @@ class SessionManager:
pass
return {"users": [], "tokens": [], "sessions": {}}
def _save_auth_data(self, auth_data: Dict[str, Any]):
"""Save auth.json data atomically."""
def _write_auth_data(self, auth_data: Dict[str, Any]):
"""Write auth.json to disk atomically. Caller MUST hold ``self._lock``."""
auth_path = self.config_dir / "auth.json"
with self._lock:
import os, tempfile
fd, tmp = tempfile.mkstemp(dir=str(self.config_dir), suffix='.tmp')
import os, tempfile
fd, tmp = tempfile.mkstemp(dir=str(self.config_dir), suffix='.tmp')
try:
with os.fdopen(fd, 'w') as f:
json.dump(auth_data, f, indent=2)
os.replace(tmp, str(auth_path))
except Exception:
try:
with os.fdopen(fd, 'w') as f:
json.dump(auth_data, f, indent=2)
os.replace(tmp, str(auth_path))
except Exception:
try:
os.unlink(tmp)
except OSError:
pass
raise
os.unlink(tmp)
except OSError:
pass
raise
def _save_auth_data(self, auth_data: Dict[str, Any]):
"""Atomically persist a full auth_data snapshot (acquires the lock).
Prefer :meth:`update_auth_data` for read-modify-write sequences: a bare
load + mutate + save is not atomic and concurrent writers can clobber
each other's changes.
"""
with self._lock:
self._write_auth_data(auth_data)
def update_auth_data(self, mutator):
"""Atomic read-modify-write of auth.json under the instance lock.
``mutator(auth_data)`` receives the freshly loaded dict, mutates it in
place, and returns ``(should_save, result)``. The file is rewritten only
when ``should_save`` is truthy; ``result`` is returned to the caller.
Holding the lock across the whole load → mutate → write cycle serializes
concurrent writers so updates can't lose each other's changes.
"""
with self._lock:
auth_data = self._load_auth_data()
should_save, result = mutator(auth_data)
if should_save:
self._write_auth_data(auth_data)
return result
def create_session(self, username: str) -> str:
"""Create a new session for a user.
......@@ -167,19 +192,17 @@ class SessionManager:
"""
session_id = secrets.token_urlsafe(32)
expires_at = utc_now() + self.session_timeout
auth_data = self._load_auth_data()
# Update sessions dict
sessions = auth_data.get("sessions", {})
sessions[session_id] = {
"username": username,
"created_at": utc_now().isoformat(),
"expires_at": expires_at.isoformat()
}
auth_data["sessions"] = sessions
self._save_auth_data(auth_data)
def _mut(auth_data):
sessions = auth_data.setdefault("sessions", {})
sessions[session_id] = {
"username": username,
"created_at": utc_now().isoformat(),
"expires_at": expires_at.isoformat(),
}
return True, None
self.update_auth_data(_mut)
# Create signed cookie value: session_id.signature
message = session_id.encode()
signature = hmac.new(self.secret, message, hashlib.sha256).hexdigest()
......@@ -204,31 +227,22 @@ class SessionManager:
expected_sig = hmac.new(self.secret, message, hashlib.sha256).hexdigest()
if not hmac.compare_digest(signature, expected_sig):
return None
# Check session in storage
auth_data = self._load_auth_data()
sessions = auth_data.get("sessions", {})
session = sessions.get(session_id)
if not session:
return None
# Check expiration
expires_at = parse_session_timestamp(session["expires_at"])
if utc_now() > expires_at:
# Clean up expired session
del sessions[session_id]
auth_data["sessions"] = sessions
self._save_auth_data(auth_data)
return None
# Extend session (sliding expiration)
new_expires = utc_now() + self.session_timeout
session["expires_at"] = new_expires.isoformat()
auth_data["sessions"] = sessions
self._save_auth_data(auth_data)
return session["username"]
# Look up, expire-check, and extend the session atomically so a
# concurrent login or token write can't clobber the sessions map.
def _mut(auth_data):
sessions = auth_data.get("sessions", {})
session = sessions.get(session_id)
if not session:
return False, None
expires_at = parse_session_timestamp(session["expires_at"])
if utc_now() > expires_at:
del sessions[session_id] # clean up expired session
return True, None
# Extend session (sliding expiration)
session["expires_at"] = (utc_now() + self.session_timeout).isoformat()
return True, session["username"]
return self.update_auth_data(_mut)
def destroy_session(self, cookie_value: Optional[str]):
"""Destroy a session."""
......@@ -239,13 +253,14 @@ class SessionManager:
session_id, _ = cookie_value.rsplit('.', 1)
except ValueError:
return
auth_data = self._load_auth_data()
sessions = auth_data.get("sessions", {})
if session_id in sessions:
del sessions[session_id]
auth_data["sessions"] = sessions
self._save_auth_data(auth_data)
def _mut(auth_data):
sessions = auth_data.get("sessions", {})
if session_id in sessions:
del sessions[session_id]
return True, None
return False, None
self.update_auth_data(_mut)
def authenticate(self, username: str, password: str) -> Optional[str]:
"""Authenticate a user and create a session.
......@@ -273,37 +288,32 @@ class SessionManager:
Returns:
True if successful, False otherwise
"""
auth_data = self._load_auth_data()
for user in auth_data.get("users", []):
if user["username"] == username:
if not verify_password(old_password, user["password_hash"]):
return False
user["password_hash"] = hash_password(new_password)
user["must_change_password"] = False
self._save_auth_data(auth_data)
return True
return False
def _mut(auth_data):
for user in auth_data.get("users", []):
if user["username"] == username:
if not verify_password(old_password, user["password_hash"]):
return False, False
user["password_hash"] = hash_password(new_password)
user["must_change_password"] = False
return True, True
return False, False
return self.update_auth_data(_mut)
def force_password_change(self, username: str, new_password: str) -> bool:
"""Force a user to change password (e.g., on first login).
Returns:
True if successful, False otherwise
"""
auth_data = self._load_auth_data()
for user in auth_data.get("users", []):
if user["username"] == username:
user["password_hash"] = hash_password(new_password)
user["must_change_password"] = False
user["last_changed_at"] = utc_now().isoformat()
self._save_auth_data(auth_data)
return True
return False
def _mut(auth_data):
for user in auth_data.get("users", []):
if user["username"] == username:
user["password_hash"] = hash_password(new_password)
user["must_change_password"] = False
user["last_changed_at"] = utc_now().isoformat()
return True, True
return False, False
return self.update_auth_data(_mut)
def get_user(self, username: str) -> Optional[Dict[str, Any]]:
"""Get user details."""
......@@ -337,28 +347,24 @@ class SessionManager:
Returns:
True if successful, False if user already exists
"""
auth_data = self._load_auth_data()
# Check if user already exists
for user in auth_data.get("users", []):
if user["username"] == username:
return False
# Find next available ID
max_id = max([u["id"] for u in auth_data.get("users", [])], default=0)
new_user = {
"id": max_id + 1,
"username": username,
"password_hash": hash_password(password),
"role": role,
"created_at": utc_now().isoformat(),
"must_change_password": False
}
auth_data["users"].append(new_user)
self._save_auth_data(auth_data)
return True
def _mut(auth_data):
# Check if user already exists
for user in auth_data.get("users", []):
if user["username"] == username:
return False, False
# Find next available ID
max_id = max([u["id"] for u in auth_data.get("users", [])], default=0)
new_user = {
"id": max_id + 1,
"username": username,
"password_hash": hash_password(password),
"role": role,
"created_at": utc_now().isoformat(),
"must_change_password": False
}
auth_data.setdefault("users", []).append(new_user)
return True, True
return self.update_auth_data(_mut)
def verify_token(self, token: str) -> bool:
"""Verify an API bearer token."""
......@@ -377,28 +383,22 @@ class SessionManager:
Returns:
True if successful, False if user doesn't exist or is last admin
"""
auth_data = self._load_auth_data()
users = auth_data.get("users", [])
# Check if last admin
admin_count = sum(1 for u in users if u.get("role") == "admin")
user = next((u for u in users if u["username"] == username), None)
if not user:
return False
if user.get("role") == "admin" and admin_count <= 1:
return False # Can't delete last admin
# Remove user
auth_data["users"] = [u for u in users if u["username"] != username]
# Remove user's sessions
sessions = auth_data.get("sessions", {})
auth_data["sessions"] = {
k: v for k, v in sessions.items()
if v["username"] != username
}
self._save_auth_data(auth_data)
return True
def _mut(auth_data):
users = auth_data.get("users", [])
# Check if last admin
admin_count = sum(1 for u in users if u.get("role") == "admin")
user = next((u for u in users if u["username"] == username), None)
if not user:
return False, False
if user.get("role") == "admin" and admin_count <= 1:
return False, False # Can't delete last admin
# Remove user
auth_data["users"] = [u for u in users if u["username"] != username]
# Remove user's sessions
sessions = auth_data.get("sessions", {})
auth_data["sessions"] = {
k: v for k, v in sessions.items()
if v["username"] != username
}
return True, True
return self.update_auth_data(_mut)
codai/admin/routes.py
View file @ f97459fc
......@@ -701,24 +701,27 @@ async def api_create_token(request: Request, username: str = Depends(require_adm
if not name:
raise HTTPException(status_code=400, detail="Token name is required")
auth_data = session_manager._load_auth_data()
# Generate token
token_id = len(auth_data.get("tokens", [])) + 1
import secrets
new_token = {
"id": token_id,
"name": name,
"token": f"sk-coderai-{secrets.token_hex(32)}",
"provider": provider,
"created_at": datetime.utcnow().isoformat() + "Z",
"last_used": None
}
auth_data.setdefault("tokens", []).append(new_token)
session_manager._save_auth_data(auth_data)
from datetime import datetime, timezone
def _mut(auth_data):
tokens = auth_data.setdefault("tokens", [])
# max+1 (not len+1) so IDs aren't reused after a deletion.
token_id = max([t.get("id", 0) for t in tokens], default=0) + 1
new_token = {
"id": token_id,
"name": name,
"token": f"sk-coderai-{secrets.token_hex(32)}",
"provider": provider,
"created_at": datetime.now(timezone.utc).isoformat(),
"last_used": None,
}
tokens.append(new_token)
return True, new_token
new_token = session_manager.update_auth_data(_mut)
return {
"token": new_token["token"],
"id": new_token["id"],
......@@ -730,16 +733,17 @@ async def api_create_token(request: Request, username: str = Depends(require_adm
@router.delete("/admin/api/tokens/{token_id}", summary="Revoke an API token")
async def api_delete_token(token_id: int, username: str = Depends(require_admin)):
"""Delete an API token."""
auth_data = session_manager._load_auth_data()
tokens = auth_data.get("tokens", [])
new_tokens = [t for t in tokens if t["id"] != token_id]
if len(new_tokens) == len(tokens):
def _mut(auth_data):
tokens = auth_data.get("tokens", [])
new_tokens = [t for t in tokens if t["id"] != token_id]
if len(new_tokens) == len(tokens):
return False, False # not found → don't write
auth_data["tokens"] = new_tokens
return True, True
if not session_manager.update_auth_data(_mut):
raise HTTPException(status_code=404, detail="Token not found")
auth_data["tokens"] = new_tokens
session_manager._save_auth_data(auth_data)
return {"success": True}
......@@ -1296,13 +1300,22 @@ async def api_model_upload(request: Request, username: str = Depends(require_adm
filename = form.get("filename", "model.gguf")
chunk_index = int(form.get("chunk_index", 0))
total_chunks = int(form.get("total_chunks", 1))
if not chunk or not hasattr(chunk, "read"):
raise HTTPException(status_code=400, detail="No file chunk provided")
# Reject path-traversal in the client-supplied filename: it must be a bare
# name that stays inside the model cache directory.
filename = os.path.basename(str(filename))
if not filename or filename in (".", "..") or "/" in filename or "\\" in filename:
raise HTTPException(status_code=400, detail="Invalid filename")
cache_dir = get_model_cache_dir()
temp_dir = tempfile.gettempdir()
upload_id = form.get("upload_id", filename)
# upload_id only ever names a temp scratch file; derive a safe slug from it
# so it can't escape temp_dir or collide via traversal either.
upload_id = os.path.basename(str(form.get("upload_id", filename))) or filename
upload_id = re.sub(r"[^A-Za-z0-9._-]", "_", upload_id)
temp_path = os.path.join(temp_dir, f"upload_{upload_id}.part")
# Append chunk
......@@ -1313,6 +1326,15 @@ async def api_model_upload(request: Request, username: str = Depends(require_adm
# If last chunk, move to final location
if chunk_index == total_chunks - 1:
final_path = os.path.join(cache_dir, filename)
# Belt-and-suspenders: ensure the resolved destination is still inside
# the cache directory before committing the upload.
if os.path.commonpath([os.path.realpath(final_path),
os.path.realpath(cache_dir)]) != os.path.realpath(cache_dir):
try:
os.remove(temp_path)
except OSError:
pass
raise HTTPException(status_code=400, detail="Invalid destination path")
os.replace(temp_path, final_path)
return {"success": True, "complete": True, "path": final_path}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment