| 1 | #!/usr/bin/env python3 |
| 2 | """Shared, dependency-light OPC package relationship validation.""" |
| 3 | |
| 4 | from __future__ import annotations |
| 5 | |
| 6 | import posixpath |
| 7 | import re |
| 8 | from pathlib import Path |
| 9 | from urllib.parse import urlsplit |
| 10 | from xml.etree import ElementTree as ET |
| 11 | |
| 12 | |
| 13 | PACKAGE_REL_NS = ( |
| 14 | "http://schemas.openxmlformats.org/package/2006/relationships" |
| 15 | ) |
| 16 | _RELATIONSHIPS_TAG = f"{{{PACKAGE_REL_NS}}}Relationships" |
| 17 | _RELATIONSHIP_TAG = f"{{{PACKAGE_REL_NS}}}Relationship" |
| 18 | _OPC_UNRESERVED = frozenset( |
| 19 | "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~" |
| 20 | ) |
| 21 | _ASCII_LOWER_TRANSLATION = str.maketrans( |
| 22 | "ABCDEFGHIJKLMNOPQRSTUVWXYZ", |
| 23 | "abcdefghijklmnopqrstuvwxyz", |
| 24 | ) |
| 25 | |
| 26 | |
| 27 | def canonical_opc_part_path(path: str) -> str | None: |
| 28 | """Return an OPC-equivalent package path key, or None when invalid.""" |
| 29 | if ( |
| 30 | not path |
| 31 | or "\\" in path |
| 32 | or "?" in path |
| 33 | or "#" in path |
| 34 | or path.endswith("/") |
| 35 | or "//" in path |
| 36 | or any(ord(char) <= 0x20 for char in path) |
| 37 | ): |
| 38 | return None |
| 39 | output: list[str] = [] |
| 40 | index = 0 |
| 41 | while index < len(path): |
| 42 | char = path[index] |
| 43 | if char != "%": |
| 44 | output.append(char) |
| 45 | index += 1 |
| 46 | continue |
| 47 | if ( |
| 48 | index + 2 >= len(path) |
| 49 | or re.fullmatch( |
| 50 | r"[0-9A-Fa-f]{2}", |
| 51 | path[index + 1:index + 3], |
| 52 | ) |
| 53 | is None |
| 54 | ): |
| 55 | return None |
| 56 | value = int(path[index + 1:index + 3], 16) |
| 57 | decoded = chr(value) |
| 58 | if value in {0, ord("/"), ord("\\")}: |
| 59 | return None |
| 60 | output.append( |
| 61 | decoded |
| 62 | if decoded in _OPC_UNRESERVED |
| 63 | else f"%{value:02X}" |
| 64 | ) |
| 65 | index += 3 |
| 66 | |
| 67 | decoded_path = "".join(output) |
| 68 | if decoded_path.rsplit("/", 1)[-1] in {".", ".."}: |
| 69 | return None |
| 70 | normalized = posixpath.normpath(decoded_path) |
| 71 | if ( |
| 72 | not normalized |
| 73 | or normalized in {".", ".."} |
| 74 | or normalized.startswith("/") |
| 75 | or normalized.startswith("../") |
| 76 | ): |
| 77 | return None |
| 78 | return normalized.translate(_ASCII_LOWER_TRANSLATION) |
| 79 | |
| 80 | |
| 81 | def _source_part_for_rels(rels_path: str) -> str | None: |
| 82 | filename = posixpath.basename(rels_path) |
| 83 | if filename == ".rels" or not filename.endswith(".rels"): |
| 84 | return None |
| 85 | source_dir = posixpath.dirname(posixpath.dirname(rels_path)) |
| 86 | source_name = filename.removesuffix(".rels") |
| 87 | return ( |
| 88 | posixpath.join(source_dir, source_name) |
| 89 | if source_dir |
| 90 | else source_name |
| 91 | ) |
| 92 | |
| 93 | |
| 94 | def resolve_internal_opc_target( |
| 95 | rels_path: str, |
| 96 | target: str, |
| 97 | ) -> str | None: |
| 98 | """Resolve one valid internal OPC Target to its canonical package key.""" |
| 99 | target_path_query = target.split("#", 1)[0] |
| 100 | if ( |
| 101 | "\\" in target |
| 102 | or "?" in target_path_query |
| 103 | or any(ord(char) <= 0x20 for char in target) |
| 104 | ): |
| 105 | return None |
| 106 | try: |
| 107 | parsed = urlsplit(target) |
| 108 | except ValueError: |
| 109 | return None |
| 110 | if parsed.scheme or parsed.netloc or parsed.query: |
| 111 | return None |
| 112 | |
| 113 | source_part = _source_part_for_rels(rels_path) |
| 114 | if parsed.path.startswith("/"): |
| 115 | resolved = parsed.path[1:] |
| 116 | elif parsed.path: |
| 117 | base_dir = posixpath.dirname(source_part) if source_part else "" |
| 118 | resolved = ( |
| 119 | posixpath.join(base_dir, parsed.path) |
| 120 | if base_dir |
| 121 | else parsed.path |
| 122 | ) |
| 123 | elif source_part and "#" in target: |
| 124 | resolved = source_part |
| 125 | else: |
| 126 | return None |
| 127 | return canonical_opc_part_path(resolved) |
| 128 | |
| 129 | |
| 130 | def verify_internal_relationships(extract_dir: Path) -> list[str]: |
| 131 | """Return invalid or dangling internal relationships in an OPC package.""" |
| 132 | package_parts: set[str] = set() |
| 133 | for path in extract_dir.rglob("*"): |
| 134 | if not path.is_file(): |
| 135 | continue |
| 136 | key = canonical_opc_part_path( |
| 137 | path.relative_to(extract_dir).as_posix() |
| 138 | ) |
| 139 | if key is not None: |
| 140 | package_parts.add(key) |
| 141 | |
| 142 | problems: list[str] = [] |
| 143 | for rels_path in sorted(extract_dir.rglob("*.rels")): |
| 144 | rels_rel = rels_path.relative_to(extract_dir).as_posix() |
| 145 | try: |
| 146 | root = ET.parse(rels_path).getroot() |
| 147 | except ET.ParseError as exc: |
| 148 | problems.append( |
| 149 | f"{rels_rel} -> <invalid relationships XML: {exc}>" |
| 150 | ) |
| 151 | continue |
| 152 | if root.tag != _RELATIONSHIPS_TAG: |
| 153 | problems.append( |
| 154 | f"{rels_rel} -> <invalid Relationships namespace>" |
| 155 | ) |
| 156 | continue |
| 157 | |
| 158 | seen_ids: set[str] = set() |
| 159 | for element in root: |
| 160 | if element.tag != _RELATIONSHIP_TAG: |
| 161 | problems.append( |
| 162 | f"{rels_rel} -> <invalid relationships child " |
| 163 | f"{element.tag!r}>" |
| 164 | ) |
| 165 | continue |
| 166 | relationship_id = (element.attrib.get("Id") or "").strip() |
| 167 | relationship_type = ( |
| 168 | element.attrib.get("Type") or "" |
| 169 | ).strip() |
| 170 | target = (element.attrib.get("Target") or "").strip() |
| 171 | target_mode = ( |
| 172 | element.attrib.get("TargetMode") or "" |
| 173 | ).strip() |
| 174 | |
| 175 | if not relationship_id: |
| 176 | problems.append(f"{rels_rel} -> <missing relationship Id>") |
| 177 | elif relationship_id in seen_ids: |
| 178 | problems.append( |
| 179 | f"{rels_rel} -> <duplicate relationship Id " |
| 180 | f"{relationship_id!r}>" |
| 181 | ) |
| 182 | else: |
| 183 | seen_ids.add(relationship_id) |
| 184 | if not relationship_type: |
| 185 | problems.append( |
| 186 | f"{rels_rel} -> <missing relationship Type>" |
| 187 | ) |
| 188 | if not target: |
| 189 | problems.append(f"{rels_rel} -> <missing Target>") |
| 190 | continue |
| 191 | if target_mode and target_mode.lower() not in { |
| 192 | "internal", |
| 193 | "external", |
| 194 | }: |
| 195 | problems.append( |
| 196 | f"{rels_rel} -> <invalid TargetMode " |
| 197 | f"{target_mode!r}>" |
| 198 | ) |
| 199 | continue |
| 200 | if target_mode.lower() == "external": |
| 201 | continue |
| 202 | |
| 203 | resolved = resolve_internal_opc_target(rels_rel, target) |
| 204 | if resolved is None: |
| 205 | problems.append( |
| 206 | f"{rels_rel} -> <invalid Target {target!r}>" |
| 207 | ) |
| 208 | elif resolved not in package_parts: |
| 209 | problems.append(f"{rels_rel} -> {resolved}") |
| 210 | return problems |
| 211 |