XLSX·XLSM을 열기 전 매크로 표시와 외부 관계를 목록으로 기록하기

#!/usr/bin/env python3
"""XLSX/XLSM 패키지 표시 목록. Office 실행·추출·네트워크 요청 없음."""
import argparse
from dataclasses import dataclass
import hashlib
import json
from pathlib import Path
import posixpath
import re
import sys
import stat
import unicodedata
from urllib.parse import urlsplit
import xml.etree.ElementTree as ET
import zipfile
import zlib

CT_NS = 'http://schemas.openxmlformats.org/package/2006/content-types'
REL_NS = 'http://schemas.openxmlformats.org/package/2006/relationships'
OFFICE = 'http://schemas.openxmlformats.org/officeDocument/2006/relationships/'
STRICT = 'http://purl.oclc.org/ooxml/officeDocument/relationships/'
VBA_REL = 'http://schemas.microsoft.com/office/2006/relationships/vbaProject'
VBA_CT = 'application/vnd.ms-office.vbaProject'
MACRO_CT = {'application/vnd.ms-excel.sheet.macroEnabled.main+xml',
            'application/vnd.ms-excel.template.macroEnabled.main+xml',
            'application/vnd.ms-excel.addin.macroEnabled.main+xml'}
CONNECTION_CT = {
    'application/vnd.openxmlformats-officedocument.spreadsheetml.connections+xml': 'connections',
    'application/vnd.openxmlformats-officedocument.spreadsheetml.externalLink+xml': 'external_link',
}
INTERESTING_REL = {VBA_REL: 'vba_relationship'}
for prefix in (OFFICE, STRICT):
    INTERESTING_REL.update({prefix + 'connections': 'connections_relationship',
                           prefix + 'externalLink': 'external_link_relationship'})
HYPERLINK_REL = {p + 'hyperlink' for p in (OFFICE, STRICT)}


@dataclass(frozen=True)
class Limits:
    input_bytes: int = 64 * 1024 * 1024
    members: int = 4096
    member_bytes: int = 32 * 1024 * 1024
    total_bytes: int = 64 * 1024 * 1024
    xml_bytes: int = 1024 * 1024
    ratio: int = 200


class StopInventory(Exception):
    def __init__(self, code, source, message, status='deferred'):
        self.code, self.source, self.message, self.status = code, source, message, status


def stop(code, source, message, status='deferred'):
    raise StopInventory(code, source, message, status)


def member_name(raw):
    """보수적 지원 범위. 모호한 이름을 고쳐 추정하지 않는다."""
    name = unicodedata.normalize('NFC', raw)
    pieces = name.rstrip('/').split('/')
    if (not name or name.startswith('/') or any(c in name for c in '\\:%\x00')
            or any(p in ('', '.', '..') for p in pieces)):
        stop('unsupported_member_path', raw, '정규 ZIP 파트 이름 지원 범위를 벗어났습니다.')
    return '/'.join(pieces)


def internal_target(owner, target):
    uri = urlsplit(target)
    if uri.scheme or uri.netloc or uri.query or uri.fragment or '\\' in target or '%' in target:
        stop('unsupported_internal_target', owner, '내부 Target URI 지원 범위를 벗어났습니다.')
    pieces = [] if target.startswith('/') else posixpath.dirname(owner).split('/')
    pieces = [p for p in pieces if p]
    for piece in target.lstrip('/').split('/'):
        if piece in ('', '.'):
            continue
        if piece == '..':
            if not pieces:
                stop('target_escapes_package', owner, '내부 Target이 패키지 루트를 벗어납니다.', 'failed')
            pieces.pop()
        else:
            pieces.append(piece)
    if not pieces:
        stop('unsupported_internal_target', owner, '빈 내부 파트 Target입니다.')
    return member_name('/'.join(pieces))


def relationship_owner(name):
    if name == '_rels/.rels':
        return ''
    folder, base = posixpath.split(name)
    if not folder.endswith('/_rels') or not base.endswith('.rels'):
        stop('unsupported_relationship_path', name, '관계파일 경로 지원 범위를 벗어났습니다.')
    return posixpath.join(folder[:-6], base[:-5])


def read_members(archive, limits):
    infos = archive.infolist()
    if len(infos) > limits.members:
        stop('too_many_members', 'ZIP', 'ZIP 항목 수 제한을 넘었습니다.')
    entries, keys, total = {}, set(), 0
    for info in infos:
        name = member_name(info.orig_filename)
        key = name.casefold()
        if key in keys:
            stop('duplicate_or_ambiguous_member', info.filename, '중복·대소문자·정규화 충돌입니다.')
        keys.add(key)
        if info.flag_bits & (1 | 64):
            stop('encrypted_zip_member', info.filename, '암호화 ZIP 항목은 읽지 않습니다.')
        if info.compress_type not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED):
            stop('unsupported_compression', info.filename, '이 예제는 Stored/Deflate만 지원합니다.')
        total += info.file_size
        if info.file_size > limits.member_bytes or total > limits.total_bytes:
            stop('member_or_total_too_large', info.filename, '압축해제 크기 제한을 넘었습니다.')
        if info.file_size > max(1, info.compress_size) * limits.ratio:
            stop('compression_ratio_limit', info.filename, '압축률 제한을 넘었습니다.')
        metadata = name == '[Content_Types].xml' or name.endswith('.rels')
        if metadata and info.file_size > limits.xml_bytes:
            stop('xml_too_large', info.filename, '메타데이터 XML 크기 제한을 넘었습니다.')
        if not info.is_dir():
            entries[name] = (info, metadata)
    # CRC 검사도 읽기로 수행한다. 내부 파일을 디스크에 추출하지 않는다.
    xml = {}
    for name, (info, metadata) in entries.items():
        chunks, size = [], 0
        with archive.open(info, 'r') as handle:
            while chunk := handle.read(65536):
                size += len(chunk)
                if size > limits.member_bytes or size > info.file_size:
                    stop('member_read_limit', name, '실제 읽기 크기가 제한·선언을 넘었습니다.')
                if metadata:
                    chunks.append(chunk)
        if size != info.file_size:
            stop('member_size_mismatch', name, 'ZIP 항목 크기 불일치입니다.', 'failed')
        if metadata:
            xml[name] = b''.join(chunks)
    return set(entries), xml


def parse_xml(data, source, expected):
    # UTF-8/ASCII만 지원. DTD·ENTITY는 파서에 보내기 전에 보류한다.
    try:
        text = data.decode('utf-8-sig')
    except UnicodeError:
        stop('unsupported_xml_encoding', source, 'UTF-8 메타데이터만 지원합니다.')
    declaration = re.match(r'\s*<\?xml\s+([^?]*)\?>', text)
    encoding = (re.search(r'''encoding\s*=\s*['"]([^'"]+)['"]''', declaration[1], re.I)
                if declaration else None)
    if encoding and encoding[1].casefold() not in ('utf-8', 'us-ascii'):
        stop('unsupported_xml_encoding', source, 'UTF-8/ASCII 선언만 지원합니다.')
    upper = text.upper()
    if '\x00' in text or '<!DOCTYPE' in upper or '<!ENTITY' in upper:
        stop('unsupported_xml_construct', source, 'DTD·ENTITY·NUL XML은 지원하지 않습니다.')
    try:
        root = ET.fromstring(text)
    except ET.ParseError:
        stop('xml_parse_failed', source, '메타데이터 XML 파싱에 실패했습니다.', 'failed')
    if root.tag != expected:
        stop('unsupported_xml_root', source, '메타데이터 루트·네임스페이스가 다릅니다.')
    return root


def collect_records(parts, xml, extension):
    records = []

    def add(kind, source, **fields):
        row = dict(kind=kind, source=source, relationship_file=None, id=None,
                   type=None, target_mode=None, target=None, part=None,
                   declared_content_type=None, declaration_kind=None,
                   exists=None, mismatch=[])
        row.update(fields)
        records.append(row)

    def presence(source, part, missing, **fields):
        add('related_part_presence', source, part=part, exists=part in parts,
            mismatch=[] if part in parts else [missing], **fields)

    if '[Content_Types].xml' not in xml:
        stop('content_types_missing', 'ZIP', '[Content_Types].xml이 없습니다.', 'failed')
    root = parse_xml(xml['[Content_Types].xml'], '[Content_Types].xml', f'{{{CT_NS}}}Types')
    seen = set()
    for item in root:
        declaration = item.tag.removeprefix(f'{{{CT_NS}}}')
        if declaration not in ('Default', 'Override'):
            stop('unsupported_content_type_element', '[Content_Types].xml', '알 수 없는 선언입니다.')
        key = item.get('PartName') if declaration == 'Override' else item.get('Extension')
        ct = item.get('ContentType')
        if not key or not ct or (declaration, key.casefold()) in seen:
            stop('invalid_content_type_declaration', '[Content_Types].xml', '누락·중복 선언입니다.', 'failed')
        seen.add((declaration, key.casefold()))
        if declaration == 'Override':
            if not key.startswith('/'):
                stop('invalid_override_part', '[Content_Types].xml',
                     'Override PartName은 /로 시작해야 합니다.', 'failed')
            matches = [internal_target('', key)]
        else:
            matches = sorted(p for p in parts if Path(p).suffix[1:].casefold() == key.casefold())
        if ct == VBA_CT:
            kind = 'vba_content_type'
        elif ct in MACRO_CT:
            kind = 'macro_enabled_content_type'
        elif ct in CONNECTION_CT:
            kind = CONNECTION_CT[ct] + '_content_type'
        else:
            kind = None
        if not kind:
            continue
        fields = dict(declared_content_type=ct, declaration_kind=declaration)
        mismatch = ['extension_macro_enabled_content_type'] if ct in MACRO_CT and extension == '.xlsx' else []
        # Default 선언 자체는 1행. 이를 적용받는 존재 파트는 별도 행이다.
        declared_part = matches[0] if declaration == 'Override' else None
        add(kind, '[Content_Types].xml', part=declared_part, target=key,
            mismatch=mismatch, **fields)
        for part in matches:
            presence('[Content_Types].xml', part, 'declared_part_missing', **fields)
    for relfile in sorted(n for n in xml if n.endswith('.rels')):
        owner = relationship_owner(relfile)
        if owner and owner not in parts:
            stop('relationship_owner_missing', relfile, '관계의 원본 파트가 없습니다.', 'failed')
        root = parse_xml(xml[relfile], relfile, f'{{{REL_NS}}}Relationships')
        ids = set()
        for item in root:
            rid, typ, target = (item.get(n) for n in ('Id', 'Type', 'Target'))
            mode = item.get('TargetMode', 'Internal')
            if (item.tag != f'{{{REL_NS}}}Relationship' or not rid or not typ
                    or not target or rid in ids or mode not in ('Internal', 'External')):
                stop('invalid_relationship', relfile, '관계 필수값·Id·TargetMode 오류입니다.', 'failed')
            ids.add(rid)
            fields = dict(relationship_file=relfile, id=rid, type=typ, target_mode=mode, target=target)
            if mode == 'External':
                kind = ('external_hyperlink_relationship' if typ in HYPERLINK_REL
                        else 'external_non_hyperlink_relationship')
                add(kind, relfile, **fields)
                # External Target은 내부 파트로 해석하거나 존재 검사를 하지 않는다.
            else:
                part = internal_target(owner, target)
                if typ in INTERESTING_REL:
                    add(INTERESTING_REL[typ], relfile, part=part, **fields)
                    presence(relfile, part, 'relationship_part_missing', **fields)
    for part in sorted(parts):
        if part == 'xl/vbaProject.bin':
            kind = 'vba_known_name'
        elif part == 'xl/connections.xml':
            kind = 'connections_known_name'
        elif part.startswith('xl/externalLinks/') and part.endswith('.xml') and '/_rels/' not in part:
            kind = 'external_link_known_name'
        else:
            continue
        add(kind, 'ZIP entry name', part=part, exists=True)
    return records


def inspect_package(path, limits=Limits()):
    path = Path(path)
    report = dict(schema_version=1, file=path.name, extension=path.suffix.lower(),
                  input_sha256=None, status='failed', indicator_count=None,
                  scope='metadata_inventory_only', records=[], issues=[])
    try:
        if not stat.S_ISREG(path.stat().st_mode):
            stop('unsupported_input_kind', path.name, '일반 파일만 지원합니다. 입력 바이트를 읽지 않았습니다.')
        with path.open('rb') as handle:
            digest = hashlib.sha256()
            while chunk := handle.read(1024 * 1024):
                digest.update(chunk)
            report['input_sha256'] = digest.hexdigest()
            size = handle.tell()
            if size > limits.input_bytes:
                stop('input_too_large', path.name, '입력 파일 크기 제한을 넘었습니다.')
            if report['extension'] not in ('.xlsx', '.xlsm'):
                stop('unsupported_extension', path.name, 'XLSX/XLSM 확장자만 지원합니다.')
            handle.seek(0)
            if handle.read(8) == bytes.fromhex('d0cf11e0a1b11ae1'):
                stop('compound_container', path.name, '암호화 OOXML 또는 구형 복합컨테이너는 지원하지 않습니다.')
            handle.seek(0)
            with zipfile.ZipFile(handle, 'r') as archive:
                parts, xml = read_members(archive, limits)
                records = collect_records(parts, xml, report['extension'])
            handle.seek(0)
            final = hashlib.sha256()
            while chunk := handle.read(1024 * 1024):
                final.update(chunk)
            if final.hexdigest() != report['input_sha256']:
                stop('input_changed', path.name, '검사 도중 입력 바이트가 변경되었습니다.', 'failed')
        for row in records:
            row.update(input_sha256=report['input_sha256'], extension=report['extension'])
        report.update(status='completed', records=records,
                      indicator_count=sum(r['kind'] != 'related_part_presence' for r in records))
    except StopInventory as error:
        report['status'] = error.status
        report['issues'].append(dict(code=error.code, source=error.source, message=error.message))
    except (OSError, zipfile.BadZipFile, EOFError, RuntimeError,
            NotImplementedError, ValueError, zlib.error) as error:
        report['issues'].append(dict(code='input_or_zip_read_failed', source=path.name,
                                     message=f'입력·ZIP 읽기 실패: {type(error).__name__}'))
    return report


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('input', type=Path, help='읽기만 할 .xlsx/.xlsm 파일')
    parser.add_argument('-o', '--output', type=Path, required=True, help='새 JSON 파일; 덮어쓰기 금지')
    args = parser.parse_args()
    try:
        # 출력 경합·동일 입력 경로도 x 모드로 기존 바이트를 보존한다.
        with args.output.open('x', encoding='utf-8') as handle:
            report = inspect_package(args.input)
            json.dump(report, handle, ensure_ascii=False, indent=2)
            handle.write('\n')
    except FileExistsError:
        print('기존 출력 파일을 보존했습니다. 새 출력 경로를 지정하세요.', file=sys.stderr)
        return 3
    except OSError as error:
        print(f'출력 저장 실패: {type(error).__name__}', file=sys.stderr)
        return 3
    print(f"{report['status']}: {args.output}")
    return 0 if report['status'] == 'completed' else 2


if __name__ == '__main__':
    raise SystemExit(main())