diff --git a/app/modules/documents/local_agent_runtime/erp_local_agent/__init__.py b/app/modules/documents/local_agent_runtime/erp_local_agent/__init__.py index 2259918..529e47e 100644 --- a/app/modules/documents/local_agent_runtime/erp_local_agent/__init__.py +++ b/app/modules/documents/local_agent_runtime/erp_local_agent/__init__.py @@ -1,2 +1,2 @@ -__version__ = "1.22.0" +__version__ = "1.22.1" AGENT_NAME = "ERP Local Agent" diff --git a/app/modules/documents/local_agent_runtime/erp_local_agent/sync.py b/app/modules/documents/local_agent_runtime/erp_local_agent/sync.py index 1501069..6c11082 100644 --- a/app/modules/documents/local_agent_runtime/erp_local_agent/sync.py +++ b/app/modules/documents/local_agent_runtime/erp_local_agent/sync.py @@ -1,7 +1,11 @@ from __future__ import annotations import shutil +import stat +import tarfile import time +import uuid +import zipfile from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -14,6 +18,166 @@ from .storage_identity import ensure_storage_identity from . import __version__ +MAX_ARCHIVE_MEMBERS = 50000 +MAX_ARCHIVE_EXTRACTED_BYTES = 20 * 1024 * 1024 * 1024 # 20 GiB hard safety ceiling. + + +def _archive_kind(path: Path) -> str | None: + """Return a supported archive kind from both filename and file signature.""" + name = path.name.lower() + wants_zip = name.endswith(".zip") + wants_tar = name.endswith((".tar", ".tar.gz", ".tgz")) + if not (wants_zip or wants_tar): + return None + if wants_zip: + if not zipfile.is_zipfile(path): + raise RuntimeError(f"Uploaded ZIP archive is invalid or damaged: {path.name}") + return "zip" + try: + if tarfile.is_tarfile(path): + return "tar" + except OSError: + pass + raise RuntimeError(f"Uploaded TAR archive is invalid or damaged: {path.name}") + + +def _archive_folder_name(filename: str) -> str: + lower = filename.lower() + if lower.endswith(".tar.gz"): + base = filename[:-7] + elif lower.endswith(".tgz"): + base = filename[:-4] + elif lower.endswith(".tar") or lower.endswith(".zip"): + base = filename.rsplit(".", 1)[0] + else: + base = Path(filename).stem + return sanitize_segment(base, "extracted") + + +def _safe_archive_member_relative(name: str) -> Path | None: + """Convert an archive member name into a safe relative local path. + + Absolute paths, drive-qualified paths and traversal components are rejected. + Individual path segments are normalised with the Local Agent's existing + filename sanitizer so extraction remains safe on Windows storage nodes. + """ + raw = str(name or "").replace("\\", "/").strip() + if not raw: + return None + if raw.startswith("/") or raw.startswith("//"): + raise RuntimeError(f"Unsafe absolute archive member path: {name}") + parts = [] + for part in raw.split("/"): + if part in ("", "."): + continue + if part == "..": + raise RuntimeError(f"Unsafe archive traversal path: {name}") + if not parts and len(part) >= 2 and part[1] == ":": + raise RuntimeError(f"Unsafe drive-qualified archive member path: {name}") + parts.append(sanitize_segment(part, "item")) + return Path(*parts) if parts else None + + +def _ensure_extract_capacity(member_count: int, total_bytes: int) -> None: + if member_count > MAX_ARCHIVE_MEMBERS: + raise RuntimeError(f"Archive contains too many entries ({member_count:,}); maximum is {MAX_ARCHIVE_MEMBERS:,}.") + if total_bytes > MAX_ARCHIVE_EXTRACTED_BYTES: + raise RuntimeError( + f"Archive expands beyond the {MAX_ARCHIVE_EXTRACTED_BYTES // (1024 ** 3)} GiB local extraction safety limit." + ) + + +def _extract_zip_safely(archive: Path, target_root: Path) -> None: + with zipfile.ZipFile(archive, "r") as zf: + infos = zf.infolist() + _ensure_extract_capacity(len(infos), sum(max(0, int(info.file_size or 0)) for info in infos)) + for info in infos: + mode = (int(info.external_attr or 0) >> 16) & 0xFFFF + if mode and stat.S_ISLNK(mode): + raise RuntimeError(f"Archive contains an unsupported symbolic link: {info.filename}") + rel = _safe_archive_member_relative(info.filename) + if rel is None: + continue + destination = (target_root / rel).resolve() + if target_root.resolve() not in destination.parents and destination != target_root.resolve(): + raise RuntimeError(f"Archive member resolves outside extraction folder: {info.filename}") + if info.is_dir(): + destination.mkdir(parents=True, exist_ok=True) + continue + destination.parent.mkdir(parents=True, exist_ok=True) + with zf.open(info, "r") as src, destination.open("wb") as dst: + shutil.copyfileobj(src, dst, length=1024 * 1024) + + +def _extract_tar_safely(archive: Path, target_root: Path) -> None: + with tarfile.open(archive, "r:*") as tf: + members = tf.getmembers() + _ensure_extract_capacity(len(members), sum(max(0, int(member.size or 0)) for member in members if member.isfile())) + for member in members: + if member.issym() or member.islnk() or member.isdev() or member.isfifo(): + raise RuntimeError(f"Archive contains an unsupported link/device member: {member.name}") + rel = _safe_archive_member_relative(member.name) + if rel is None: + continue + destination = (target_root / rel).resolve() + if target_root.resolve() not in destination.parents and destination != target_root.resolve(): + raise RuntimeError(f"Archive member resolves outside extraction folder: {member.name}") + if member.isdir(): + destination.mkdir(parents=True, exist_ok=True) + continue + if not member.isfile(): + continue + source = tf.extractfile(member) + if source is None: + raise RuntimeError(f"Could not read archive member: {member.name}") + destination.parent.mkdir(parents=True, exist_ok=True) + with source, destination.open("wb") as dst: + shutil.copyfileobj(source, dst, length=1024 * 1024) + + +def _preferred_extracted_path(extracted_root: Path) -> Path: + """Use a single wrapper directory directly; otherwise return extraction root.""" + try: + children = list(extracted_root.iterdir()) + except OSError: + return extracted_root + if len(children) == 1 and children[0].is_dir(): + return children[0] + return extracted_root + + +def _extract_archive_to_local_storage(archive: Path) -> Path | None: + """Safely extract supported archives next to the stored source archive. + + The original archive is intentionally retained. Existing secure download and + version/hash behaviour therefore remains unchanged, while users can work from + the extracted directory in branch-local storage. + """ + kind = _archive_kind(archive) + if kind is None: + return None + final_root = archive.parent / _archive_folder_name(archive.name) + temp_root = archive.parent / f".{final_root.name}.extracting-{uuid.uuid4().hex[:10]}" + if temp_root.exists(): + shutil.rmtree(temp_root, ignore_errors=True) + temp_root.mkdir(parents=True, exist_ok=False) + try: + if kind == "zip": + _extract_zip_safely(archive, temp_root) + else: + _extract_tar_safely(archive, temp_root) + if final_root.exists(): + if final_root.is_dir(): + shutil.rmtree(final_root) + else: + final_root.unlink() + temp_root.replace(final_root) + except Exception: + shutil.rmtree(temp_root, ignore_errors=True) + raise + return _preferred_extracted_path(final_root) + + class StorageAgent: def __init__(self, config: AgentConfig, client: ERPClient, db: LocalDB, logger): self.config = config @@ -94,17 +258,49 @@ class StorageAgent: response = self.client.download_storage_job(job_id) file_size, digest = write_stream_to_file(response.raw, destination) + final_relative_path = relative_path + try: + extracted_path = _extract_archive_to_local_storage(destination) + if extracted_path is not None: + final_relative_path = extracted_path.resolve().relative_to(self.config.storage_root.resolve()) + self.logger.info( + "Archive storage job=%s extracted safely to %s", + job_id, + extracted_path, + ) + except Exception as exc: + error_payload = { + "status": "FAILED", + "success": False, + "sha256_hash": digest, + "file_hash": digest, + "file_size": file_size, + "local_relative_path": str(relative_path).replace("\\", "/"), + "error": f"Archive extraction failed: {type(exc).__name__}: {exc}", + } + try: + self.client.acknowledge_storage_job(job_id, error_payload) + finally: + self.db.record_event("ERROR", "archive_extraction_failed", error_payload["error"]) + raise + + final_relative_text = str(final_relative_path).replace("\\", "/") payload = { "status": "COMPLETED", + "success": True, "sha256_hash": digest, "file_hash": digest, "file_size": file_size, - "local_relative_path": str(relative_path).replace("\\", "/"), + "local_relative_path": final_relative_text, + "local_final_path": final_relative_text, "stored_at_utc": datetime.now(timezone.utc).isoformat(), } self.client.acknowledge_storage_job(job_id, payload) - self.db.record_storage_job(str(job_id), str(relative_path).replace("\\", "/"), digest, file_size) - self.logger.info("Storage job completed job=%s size=%s sha256=%s", job_id, file_size, digest) + self.db.record_storage_job(str(job_id), final_relative_text, digest, file_size) + self.logger.info( + "Storage job completed job=%s size=%s sha256=%s final_path=%s", + job_id, file_size, digest, final_relative_text, + ) def process_download_requests(self) -> None: try: diff --git a/app/modules/services/engagement_resources.py b/app/modules/services/engagement_resources.py index e559f5a..4f3a762 100644 --- a/app/modules/services/engagement_resources.py +++ b/app/modules/services/engagement_resources.py @@ -6,6 +6,7 @@ from typing import Any, Iterable from sqlalchemy import select +from app.modules.documents.models import DocumentStorageJob from app.modules.documents.services import client_folder_parts, get_active_storage_node_for_branch, sanitize_segment from app.modules.services.models import FirmServiceTaskTemplate, ServiceTaskCategory @@ -41,6 +42,25 @@ def _join_local_path(root: str, relative: str) -> str: return root + "/" + relative.lstrip("/") +def _latest_local_final_path(db, version: Any) -> str: + """Return the branch-agent acknowledged final path for a document version. + + For ordinary files this is the file path. For ZIP/TAR uploads handled by Local + Agent 1.22.1+ this is the extracted directory path, while the original archive + continues to remain the versioned source file for secure download/history. + """ + version_id = int(getattr(version, "id", 0) or 0) + if not version_id: + return "" + job = db.execute( + select(DocumentStorageJob).where( + DocumentStorageJob.version_id == version_id, + DocumentStorageJob.status == "completed", + ).order_by(DocumentStorageJob.completed_at_utc.desc(), DocumentStorageJob.id.desc()).limit(1) + ).scalar_one_or_none() + return str(getattr(job, "local_final_path", "") or "").strip() if job else "" + + def _task_category_options(db, engagement, tasks: list[Any]) -> list[dict[str, Any]]: """Return the service's real task-category master mapped to this engagement's tasks. @@ -203,6 +223,8 @@ def build_engagement_resource_context(db, engagement, tasks: Iterable[Any], docu versions = list(getattr(doc, "versions", None) or []) latest = versions[0] if versions else None latest_relative_path = str(getattr(latest, "local_relative_path", "") or "") if latest else "" + local_final_path = _latest_local_final_path(db, latest) if latest else "" + displayed_relative_path = local_final_path or latest_relative_path tally_year = _tally_document_year(doc) row = { "id": int(getattr(doc, "id", 0) or 0), @@ -213,6 +235,8 @@ def build_engagement_resource_context(db, engagement, tasks: Iterable[Any], docu "task_instance_id": getattr(doc, "task_instance_id", None), "latest_filename": str(getattr(latest, "original_filename", "") or "") if latest else "", "latest_relative_path": latest_relative_path, + "local_final_path": local_final_path, + "displayed_relative_path": displayed_relative_path, "storage_status": str(getattr(latest, "storage_status", "") or "") if latest else "", } document_options.append(row) @@ -220,7 +244,7 @@ def build_engagement_resource_context(db, engagement, tasks: Iterable[Any], docu tally_data_files.append({ **row, "financial_year": tally_year, - "local_path": _join_local_path(storage_root_path, latest_relative_path) if latest_relative_path else "", + "local_path": _join_local_path(storage_root_path, displayed_relative_path) if displayed_relative_path else "", }) tally_data_files.sort(key=lambda row: row.get("financial_year", ""), reverse=True)