125 lines
3.3 KiB
Python
125 lines
3.3 KiB
Python
|
"""Per-user vault file store — mirrors the client vault layout."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import time
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from .config import settings
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class VaultFileEntry:
|
||
|
|
path: str
|
||
|
|
size: int
|
||
|
|
mtime: float
|
||
|
|
sha256: str
|
||
|
|
|
||
|
|
def to_dict(self) -> dict:
|
||
|
|
return {
|
||
|
|
"path": self.path,
|
||
|
|
"size": self.size,
|
||
|
|
"mtime": self.mtime,
|
||
|
|
"sha256": self.sha256,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def vault_root(user_id: str) -> Path:
|
||
|
|
root = Path(settings.vault_path) / user_id / "files"
|
||
|
|
root.mkdir(parents=True, exist_ok=True)
|
||
|
|
return root
|
||
|
|
|
||
|
|
|
||
|
|
def _safe_relpath(rel: str) -> str:
|
||
|
|
"""Normalize and reject path escape attempts."""
|
||
|
|
cleaned = rel.replace("\\", "/").lstrip("/")
|
||
|
|
if not cleaned or cleaned.startswith("..") or "/../" in f"/{cleaned}/":
|
||
|
|
raise ValueError(f"invalid vault path: {rel!r}")
|
||
|
|
parts = Path(cleaned).parts
|
||
|
|
if ".." in parts:
|
||
|
|
raise ValueError(f"invalid vault path: {rel!r}")
|
||
|
|
return "/".join(parts)
|
||
|
|
|
||
|
|
|
||
|
|
def resolve_path(user_id: str, rel: str) -> Path:
|
||
|
|
rel_n = _safe_relpath(rel)
|
||
|
|
full = (vault_root(user_id) / rel_n).resolve()
|
||
|
|
root = vault_root(user_id).resolve()
|
||
|
|
if not str(full).startswith(str(root) + os.sep) and full != root:
|
||
|
|
raise ValueError(f"path escapes vault: {rel!r}")
|
||
|
|
return full
|
||
|
|
|
||
|
|
|
||
|
|
def _sha256_file(path: Path) -> str:
|
||
|
|
h = hashlib.sha256()
|
||
|
|
with path.open("rb") as f:
|
||
|
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
||
|
|
h.update(chunk)
|
||
|
|
return h.hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def build_manifest(user_id: str) -> dict:
|
||
|
|
root = vault_root(user_id)
|
||
|
|
entries: list[VaultFileEntry] = []
|
||
|
|
for path in sorted(root.rglob("*")):
|
||
|
|
if not path.is_file():
|
||
|
|
continue
|
||
|
|
if path.name.startswith("."):
|
||
|
|
continue
|
||
|
|
rel = path.relative_to(root).as_posix()
|
||
|
|
st = path.stat()
|
||
|
|
entries.append(
|
||
|
|
VaultFileEntry(
|
||
|
|
path=rel,
|
||
|
|
size=st.st_size,
|
||
|
|
mtime=st.st_mtime,
|
||
|
|
sha256=_sha256_file(path),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
return {
|
||
|
|
"user_id": user_id,
|
||
|
|
"generated_at": time.time(),
|
||
|
|
"files": [e.to_dict() for e in entries],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def write_file(user_id: str, rel: str, data: bytes) -> VaultFileEntry:
|
||
|
|
path = resolve_path(user_id, rel)
|
||
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
||
|
|
tmp.write_bytes(data)
|
||
|
|
tmp.replace(path)
|
||
|
|
st = path.stat()
|
||
|
|
return VaultFileEntry(
|
||
|
|
path=_safe_relpath(rel),
|
||
|
|
size=st.st_size,
|
||
|
|
mtime=st.st_mtime,
|
||
|
|
sha256=_sha256_file(path),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def read_file(user_id: str, rel: str) -> bytes:
|
||
|
|
path = resolve_path(user_id, rel)
|
||
|
|
if not path.is_file():
|
||
|
|
raise FileNotFoundError(rel)
|
||
|
|
return path.read_bytes()
|
||
|
|
|
||
|
|
|
||
|
|
def delete_file(user_id: str, rel: str, *, tombstone: bool = True) -> None:
|
||
|
|
path = resolve_path(user_id, rel)
|
||
|
|
if not path.exists():
|
||
|
|
raise FileNotFoundError(rel)
|
||
|
|
if tombstone:
|
||
|
|
tomb = path.parent / f".tombstone-{path.name}"
|
||
|
|
meta = {
|
||
|
|
"path": _safe_relpath(rel),
|
||
|
|
"deleted_at": time.time(),
|
||
|
|
"sha256": _sha256_file(path) if path.is_file() else None,
|
||
|
|
}
|
||
|
|
tomb.write_text(json.dumps(meta), encoding="utf-8")
|
||
|
|
path.unlink()
|