""" Shared storage layer — zone metadata and path utilities. Single Responsibility: only handles metadata persistence and path resolution. Dependency Inversion: routers depend on these abstractions instead of importing each other. """ import json import os from pathlib import Path from config import DATA_DIR, ZONES_META, ZONE_NAME_PATTERN def load_meta() -> dict: """Load zones metadata from JSON file.""" if ZONES_META.exists(): return json.loads(ZONES_META.read_text(encoding="utf-8")) return {} def save_meta(meta: dict): """Save zones metadata to JSON file.""" ZONES_META.write_text(json.dumps(meta, indent=2, default=str), encoding="utf-8") def validate_zone_name(name: str): """Validate zone name format. Raises ValueError if invalid.""" if not ZONE_NAME_PATTERN.match(name): raise ValueError("Tên zone chỉ chứa a-z, A-Z, 0-9, _, - (tối đa 50 ký tự)") def get_zone_path(name: str) -> Path: """Get the filesystem path for a zone, validating it exists.""" validate_zone_name(name) zone_path = DATA_DIR / name if not zone_path.is_dir(): raise ValueError(f"Zone '{name}' không tồn tại") return zone_path def safe_path(zone_path: Path, rel_path: str) -> Path: """Resolve a relative path within a zone, preventing path traversal.""" target = (zone_path / rel_path).resolve() zone_resolved = zone_path.resolve() if target != zone_resolved and not str(target).startswith(str(zone_resolved) + os.sep): raise ValueError("Truy cập ngoài zone không được phép") return target def check_zone_owner(zone_name: str, user_sub: str, user_role: str): """Check that the user owns the zone. Admins can access all zones.""" if user_role == "admin": return meta = load_meta() info = meta.get(zone_name) if not info: raise ValueError(f"Zone '{zone_name}' không tồn tại") if info.get("owner_id") != user_sub: raise ValueError("Bạn không có quyền truy cập zone này")