source_binding.py

"""Validate the generated source binding against the mounted Cosmos payload."""
import hashlib
import json
import os
import re
from pathlib import Path, PurePosixPath


def mounted_files(root, declared):
    """Inventory payload; allow only CPython caches for declared Python modules."""
    result = set()
    def fail(error):
        raise error
    for directory, directories, filenames in os.walk(root, followlinks=False, onerror=fail):
        for name in directories + filenames:
            target = Path(directory) / name
            if target.is_symlink() or getattr(target, 'is_junction', lambda: False)():
                raise ValueError('mounted link')
        for name in filenames:
            target = Path(directory) / name
            if not target.is_file():
                raise ValueError('nonregular mounted entry')
            relative = target.relative_to(root)
            if relative.as_posix() == 'COSMOS_SOURCE_BINDING.json':
                continue
            if relative.parent.name == '__pycache__':
                match = re.fullmatch(r'(.+)\.cpython-[0-9]+(?:\.opt-[012])?\.pyc', name)
                source = relative.parent.parent / ((match[1] if match else '') + '.py')
                if match and source.as_posix() in declared:
                    continue
            result.add(relative.as_posix())
    return result


def bound_source(directory):
    root = Path(directory).resolve()
    try:
        binding = json.loads((root / 'COSMOS_SOURCE_BINDING.json').read_text(encoding='utf-8'))
        if not isinstance(binding, dict):
            raise ValueError('binding object required')
        if binding['schema'] != 'szl.cosmos-source/v1' or binding['repository'] != 'szl-holdings/anatomy' or binding['source_path'] != 'spaces/cosmos' or binding['destination'] != 'betterwithage/cosmos':
            raise ValueError('binding identity')
        if not re.fullmatch('[0-9a-f]{40}', binding['commit']):
            raise ValueError('source commit')
        files = binding['files']
        if not isinstance(files, dict) or not {'server.py', 'catalog.py', 'source_binding.py', 'index.html', 'vendor/three.module.min.js'}.issubset(files):
            raise ValueError('incomplete binding')
        if mounted_files(root, files) != set(files):
            raise ValueError('mounted file set mismatch')
        for name, digest in files.items():
            path = PurePosixPath(name)
            if not name or path.as_posix() != name or path.is_absolute() or '..' in path.parts or ':' in name or '\\' in name or not re.fullmatch('[0-9a-f]{64}', digest):
                raise ValueError('invalid file entry')
            target = root.joinpath(*path.parts)
            if target.is_symlink() or not target.resolve().is_relative_to(root) or hashlib.sha256(target.read_bytes()).hexdigest() != digest:
                raise ValueError('payload mismatch')
        return {'repository': binding['repository'], 'commit': binding['commit'], 'path': binding['source_path'], 'relation': 'mounted-files-match-publisher-binding', 'state': 'SOURCE_BOUND', 'evidence_url': 'https://github.com/' + binding['repository'] + '/tree/' + binding['commit'] + '/' + binding['source_path']}, 'SOURCE_BOUND_LOCAL_BYTES'
    except (OSError, ValueError, TypeError, KeyError):
        return None, 'UNKNOWN_SOURCE_RELATION'