Extract uploaded archives into local engagement storage

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"

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)
 
-        payload = {
-            "status": "COMPLETED",
+        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": 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:

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)