Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 26 additions & 1 deletion sourceosctl/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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)
Expand Down
116 changes: 111 additions & 5 deletions sourceosctl/commands/office.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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"
Expand Down Expand Up @@ -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],
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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))
Expand All @@ -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,
},
Expand Down
85 changes: 80 additions & 5 deletions sourceosctl/commands/ooxml.py
Original file line number Diff line number Diff line change
@@ -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)
Expand Down Expand Up @@ -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'''<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<w:document xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main">
Expand Down
80 changes: 80 additions & 0 deletions tests/test_office_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
Loading