diff --git a/sourceosctl/cli.py b/sourceosctl/cli.py index 81c6c01..3e05fb7 100644 --- a/sourceosctl/cli.py +++ b/sourceosctl/cli.py @@ -393,7 +393,7 @@ def add_office_common(p): "--execute", action="store_true", default=False, - help="Write txt/md/json artifacts only; Office binary generation remains disabled", + help="Write txt/md/json/docx/xlsx/pptx artifacts under guarded local execution", ) office_generate_p.add_argument( "--policy-ok", @@ -440,6 +440,31 @@ def add_office_common(p): ) office_convert_p.set_defaults(func=office.convert) + office_validate_p = office_sub.add_parser( + "validate", help="Run Office artifact quality gates" + ) + office_validate_p.add_argument("path", help="Path to Office artifact file") + office_validate_p.add_argument("--format", default=None, help="Override detected artifact format") + office_validate_p.add_argument( + "--roundtrip", + action="store_true", + default=False, + help="Attempt a LibreOffice PDF round-trip validation when available", + ) + office_validate_p.add_argument( + "--policy-ok", + action="store_true", + default=False, + help="Confirm Policy Fabric/operator approval for round-trip execution", + ) + office_validate_p.add_argument( + "--workroom-id", default="workroom-local-default", help="Professional Workroom id" + ) + office_validate_p.add_argument( + "--evidence-out", default=None, help="Optional path to write OfficeArtifactEvidence JSON" + ) + office_validate_p.set_defaults(func=office.validate) + office_inspect_p = office_sub.add_parser("inspect", help="Inspect an Office artifact file") office_inspect_p.add_argument("path", help="Path to Office artifact file") office_inspect_p.set_defaults(func=office.inspect) diff --git a/sourceosctl/commands/office.py b/sourceosctl/commands/office.py index 6c8a841..8462640 100644 --- a/sourceosctl/commands/office.py +++ b/sourceosctl/commands/office.py @@ -1,9 +1,9 @@ """office command helpers. This module implements SourceOS Office Plane planning plus guarded local -execution. Dry-run remains the default. File-writing behavior is available -only behind --execute --policy-ok, writes only to explicit output roots, and -emits OfficeArtifactEvidence-compatible JSON. +execution and quality gates. Dry-run remains the default. File-writing +behavior is available only behind --execute --policy-ok, writes only to +explicit output roots, and emits OfficeArtifactEvidence-compatible JSON. """ from __future__ import annotations @@ -17,11 +17,16 @@ import shutil import subprocess import sys +import tempfile from pathlib import Path from typing import Any, Dict, Optional from sourceosctl.commands.office_runtime_contracts import build_office_runtime_contracts -from sourceosctl.commands.ooxml import OOXML_GENERATION_FORMATS, write_ooxml_artifact +from sourceosctl.commands.ooxml import ( + OOXML_GENERATION_FORMATS, + validate_ooxml_artifact, + write_ooxml_artifact, +) DEFAULT_WORKROOM_ID = "workroom-local-default" @@ -228,6 +233,16 @@ def _artifact_output_path(args, fmt: str) -> Path: return Path(_expand(getattr(args, "output_root", DEFAULT_OUTPUT_ROOT))) / f"{_safe_slug(title)}.{fmt}" +def _artifact_type_for_format(fmt: str) -> str: + if fmt in {"xlsx", "ods", "csv"}: + return "spreadsheet" + if fmt in {"pptx", "odp"}: + return "slide-deck" + if fmt == "pdf": + return "pdf" + return "document" + + def _build_evidence( *, plan: Dict[str, Any], @@ -398,7 +413,8 @@ def generate(args) -> int: workroom_id=payload["officeArtifact"]["workroomId"], artifact_id=payload["officeArtifact"]["artifactId"], ) - notes = "sourceosctl guarded minimal OOXML Office Plane artifact generation" + structural = validate_ooxml_artifact(output_path, fmt) + notes = f"sourceosctl guarded minimal OOXML Office Plane artifact generation; structuralValid={structural['valid']}" evidence = _build_evidence( plan=payload, @@ -490,6 +506,95 @@ def convert(args) -> int: return _print_json(result) if status == "success" else (_print_json(result) or 1) +def validate(args) -> int: + """Validate an Office artifact with structural and optional round-trip gates.""" + path = Path(_expand(args.path)) + fmt = getattr(args, "format", None) or path.suffix.lower().lstrip(".") + structural = validate_ooxml_artifact(path, fmt) if fmt in OOXML_GENERATION_FORMATS else { + "kind": "OOXMLStructuralValidation", + "format": fmt, + "path": str(path), + "valid": path.exists() and path.is_file(), + "requiredParts": [], + "missingParts": [], + "xmlErrors": [] if path.exists() and path.is_file() else ["artifact does not exist or is not a file"], + "zipEntries": [], + } + + roundtrip = { + "requested": bool(getattr(args, "roundtrip", False)), + "executed": False, + "available": False, + "status": "skipped", + "outputRef": None, + "error": None, + } + if getattr(args, "roundtrip", False): + if not getattr(args, "policy_ok", False): + print("error: office validate --roundtrip requires --policy-ok", file=sys.stderr) + return 1 + lo = _libreoffice_path() + roundtrip["available"] = lo is not None + if not lo: + roundtrip["status"] = "blocked" + roundtrip["error"] = "LibreOffice/soffice not found on PATH" + elif not path.exists() or not path.is_file(): + roundtrip["status"] = "blocked" + roundtrip["error"] = "artifact does not exist or is not a file" + else: + with tempfile.TemporaryDirectory() as tmpdir: + cmd = [lo, "--headless", "--convert-to", "pdf", "--outdir", tmpdir, str(path)] + completed = subprocess.run(cmd, check=False, capture_output=True, text=True, timeout=180) + expected = Path(tmpdir) / f"{path.stem}.pdf" + roundtrip["executed"] = True + roundtrip["status"] = "success" if completed.returncode == 0 and expected.exists() else "failure" + roundtrip["outputRef"] = str(expected.name) if expected.exists() else None + if completed.returncode != 0: + roundtrip["error"] = f"stdout={completed.stdout[-200:]!r}; stderr={completed.stderr[-200:]!r}" + + status = "success" if structural["valid"] and roundtrip["status"] in {"skipped", "success"} else "failure" + plan_args = type( + "Args", + (), + { + "artifact_type": _artifact_type_for_format(fmt), + "format": fmt, + "title": path.stem or "Office Artifact", + "workroom_id": getattr(args, "workroom_id", DEFAULT_WORKROOM_ID), + "output_root": DEFAULT_OUTPUT_ROOT, + "backend": "libreoffice", + "mode": "local-headless", + "execute": False, + "downloads_root": DEFAULT_DOWNLOADS_ROOT, + "template_root": DEFAULT_TEMPLATE_ROOT, + }, + )() + plan = _artifact_plan(plan_args, "validate", format_override=fmt) + evidence = _build_evidence( + plan=plan, + operation="analyze", + status=status, + output_path=path if path.exists() and path.is_file() else None, + source_refs=[_redact_home(str(path)) or str(path)], + notes=f"structuralValid={structural['valid']}; roundtripStatus={roundtrip['status']}", + ) + evidence_out = getattr(args, "evidence_out", None) + if evidence_out: + _write_json(evidence_out, evidence) + + result = { + "type": "OfficeQualityGateResult", + "status": status, + "format": fmt, + "path": _redact_home(str(path)), + "structural": structural, + "roundtrip": roundtrip, + "evidenceOut": _redact_home(evidence_out) if evidence_out else None, + "evidence": None if evidence_out else evidence, + } + return _print_json(result) if status == "success" else (_print_json(result) or 1) + + def inspect(args) -> int: """Inspect an Office artifact file without modifying it.""" path = Path(_expand(args.path)) @@ -512,6 +617,7 @@ def inspect(args) -> int: "supportedFormat": suffix in SUPPORTED_FORMATS, "mimeType": mime_type, "sha256": _sha256(path), + "qualityGate": validate_ooxml_artifact(path, suffix) if suffix in OOXML_GENERATION_FORMATS else None, "contracts": { "officeArtifactSchema": OFFICE_ARTIFACT_SCHEMA, }, diff --git a/sourceosctl/commands/ooxml.py b/sourceosctl/commands/ooxml.py index 04a28dc..b40f851 100644 --- a/sourceosctl/commands/ooxml.py +++ b/sourceosctl/commands/ooxml.py @@ -1,21 +1,47 @@ -"""Minimal OOXML artifact builders for SourceOS Office Plane. +"""Minimal OOXML artifact builders and validators for SourceOS Office Plane. These helpers intentionally use only Python's standard library. They create small, deterministic-enough DOCX/XLSX/PPTX ZIP packages for guarded local -artifact generation. They are not a replacement for LibreOffice, Collabora, -ONLYOFFICE, or a full template engine; they are the safe local bootstrap path -for simple agent-authored workroom artifacts. +artifact generation and validate required OOXML package structure for quality +gates. They are not a replacement for LibreOffice, Collabora, ONLYOFFICE, or a +full template engine; they are the safe local bootstrap path for simple +agent-authored workroom artifacts. """ from __future__ import annotations +import xml.etree.ElementTree as ET from html import escape from pathlib import Path -from zipfile import ZIP_DEFLATED, ZipFile +from zipfile import BadZipFile, ZIP_DEFLATED, ZipFile OOXML_GENERATION_FORMATS = {"docx", "xlsx", "pptx"} +REQUIRED_PARTS = { + "docx": ["[Content_Types].xml", "_rels/.rels", "word/document.xml"], + "xlsx": [ + "[Content_Types].xml", + "_rels/.rels", + "xl/workbook.xml", + "xl/_rels/workbook.xml.rels", + "xl/worksheets/sheet1.xml", + ], + "pptx": [ + "[Content_Types].xml", + "_rels/.rels", + "ppt/presentation.xml", + "ppt/_rels/presentation.xml.rels", + "ppt/slides/slide1.xml", + ], +} + +ROOT_PARTS = { + "docx": "word/document.xml", + "xlsx": "xl/workbook.xml", + "pptx": "ppt/presentation.xml", +} + def _xml(text: str) -> str: return escape(text, quote=True) @@ -57,6 +83,55 @@ def write_ooxml_artifact( raise ValueError(f"unsupported OOXML generation format: {fmt}") +def validate_ooxml_artifact(path: Path, fmt: str | None = None) -> dict: + """Validate basic OOXML ZIP/package structure. + + This is a structural quality gate, not a complete ECMA-376 validator. It + checks that the artifact is a ZIP package, required parts exist, and XML + parts parse successfully. + """ + fmt = fmt or path.suffix.lower().lstrip(".") + result = { + "kind": "OOXMLStructuralValidation", + "format": fmt, + "path": str(path), + "valid": False, + "requiredParts": REQUIRED_PARTS.get(fmt, []), + "missingParts": [], + "xmlErrors": [], + "zipEntries": [], + } + + if fmt not in REQUIRED_PARTS: + result["xmlErrors"].append(f"unsupported OOXML format: {fmt}") + return result + + if not path.exists() or not path.is_file(): + result["xmlErrors"].append("artifact does not exist or is not a file") + return result + + try: + with ZipFile(path, "r") as zf: + names = sorted(zf.namelist()) + result["zipEntries"] = names + for required in REQUIRED_PARTS[fmt]: + if required not in names: + result["missingParts"].append(required) + for name in names: + if not name.endswith(".xml") and not name.endswith(".rels"): + continue + try: + ET.fromstring(zf.read(name)) + except ET.ParseError as exc: + result["xmlErrors"].append(f"{name}: {exc}") + except BadZipFile: + result["xmlErrors"].append("artifact is not a valid ZIP package") + return result + + result["valid"] = not result["missingParts"] and not result["xmlErrors"] + return result + + def _write_docx(*, path: Path, title: str, workroom_id: str, artifact_id: str) -> None: document = f''' diff --git a/tests/test_office_cli.py b/tests/test_office_cli.py index 7e67545..e8920c0 100644 --- a/tests/test_office_cli.py +++ b/tests/test_office_cli.py @@ -216,6 +216,86 @@ def test_office_generate_execute_writes_pptx_package(self): ["[Content_Types].xml", "_rels/.rels", "ppt/presentation.xml", "ppt/slides/slide1.xml"], ) + def test_office_validate_accepts_generated_docx_and_writes_evidence(self): + with tempfile.TemporaryDirectory() as tmpdir: + output_path = os.path.join(tmpdir, "safe-doc.docx") + evidence_path = os.path.join(tmpdir, "evidence", "validate.json") + rc = main([ + "office", + "generate", + "--execute", + "--policy-ok", + "--artifact-type", + "document", + "--format", + "docx", + "--title", + "Safe Doc", + "--output-root", + tmpdir, + ]) + self.assertEqual(rc, 0) + rc = main([ + "office", + "validate", + output_path, + "--evidence-out", + evidence_path, + ]) + self.assertEqual(rc, 0) + with open(evidence_path, "r", encoding="utf-8") as handle: + evidence = json.load(handle) + self.assertEqual(evidence["kind"], "OfficeArtifactEvidence") + self.assertEqual(evidence["operation"], "analyze") + self.assertEqual(evidence["status"], "success") + + def test_office_validate_rejects_invalid_docx_zip(self): + with tempfile.NamedTemporaryFile(suffix=".docx", mode="w", delete=False) as handle: + handle.write("not a zip") + tmp_path = handle.name + try: + self.assertEqual(main(["office", "validate", tmp_path]), 1) + finally: + os.unlink(tmp_path) + + def test_office_validate_roundtrip_requires_policy_ok(self): + with tempfile.TemporaryDirectory() as tmpdir: + rc = main([ + "office", + "generate", + "--execute", + "--policy-ok", + "--artifact-type", + "document", + "--format", + "docx", + "--title", + "Safe Doc", + "--output-root", + tmpdir, + ]) + self.assertEqual(rc, 0) + self.assertEqual(main(["office", "validate", os.path.join(tmpdir, "safe-doc.docx"), "--roundtrip"]), 1) + + def test_office_inspect_includes_quality_gate_for_ooxml(self): + with tempfile.TemporaryDirectory() as tmpdir: + rc = main([ + "office", + "generate", + "--execute", + "--policy-ok", + "--artifact-type", + "slide-deck", + "--format", + "pptx", + "--title", + "Safe Deck", + "--output-root", + tmpdir, + ]) + self.assertEqual(rc, 0) + self.assertEqual(main(["office", "inspect", os.path.join(tmpdir, "safe-deck.pptx")]), 0) + def test_office_generate_execute_rejects_whole_home_output_root(self): args = _office_args( execute=True,