파일 선택 확인과 저장 확인을 나누는 작은 모듈

"""Independent synthetic byte/selection receipts; no filesystem or network access."""
from dataclasses import dataclass
from hashlib import sha256
from uuid import uuid4
import re

MAX_BYTES = 65536


def payload(value):
    if type(value) is not bytes or len(value) > MAX_BYTES:
        raise ValueError('BYTES_REQUIRED_WITHIN_64_KIB')
    return value


def filename(value):
    if (type(value) is not str or not value or len(value) > 140
            or value != value.strip() or value.endswith('.')
            or value in ('.', '..') or re.search(r'[<>:"/\\|?*\x00-\x1f]', value)
            or re.match(r'^(con|prn|aux|nul|com[1-9]|lpt[1-9])(?:\.|$)', value, re.I)):
        raise ValueError('BASENAME_REQUIRED')
    return value


@dataclass(frozen=True)
class Scope:
    profile: str
    project: str

    def __post_init__(self):
        if any(type(x) is not str or not x.strip() or len(x) > 80
               for x in (self.profile, self.project)):
            raise ValueError('SCOPE_REQUIRED')


class Selection:
    """Immutable selected snapshots plus mutable synthetic current bytes."""
    def __init__(self):
        self._rows = {}

    def add(self, scope, name, data):
        if type(scope) is not Scope:
            raise ValueError('SCOPE_REQUIRED')
        name, data = filename(name), payload(data)
        for key, row in self._rows.items():
            if row['scope'] == scope and row['name'] == name and row['original'] == data and row['current'] == data:
                return key
        if sum(r['scope'] == scope for r in self._rows.values()) >= 20:
            raise ValueError('SELECTION_LIMIT')
        key = str(uuid4())
        self._rows[key] = {'scope': scope, 'name': name, 'original': data, 'current': data}
        return key

    def _row(self, scope, handle):
        row = self._rows.get(handle)
        if type(scope) is not Scope or row is None or row['scope'] != scope:
            raise ValueError('HANDLE_DENIED')
        return row

    def change_synthetic_bytes(self, scope, handle, data):
        row = self._row(scope, handle)
        row['current'] = payload(data)

    def revoke(self, scope, handle):
        self._row(scope, handle)
        del self._rows[handle]

    def describe(self, scope, allowed, handles):
        if type(allowed) is not list or type(handles) is not list or not handles:
            raise ValueError('NONEMPTY_HANDLE_LIST_REQUIRED')
        if any(type(x) is not str for x in allowed + handles) or len(set(handles)) != len(handles):
            raise ValueError('DISTINCT_STRING_HANDLES_REQUIRED')
        result = []
        for key in handles:
            if key not in allowed:
                raise ValueError('HANDLE_DENIED')
            row = self._row(scope, key)
            if row['current'] != row['original']:
                raise ValueError('SELECTED_BYTES_CHANGED')
            data = row['current']
            result.append({'id': key, 'name': row['name'], 'size': len(data), 'sha256': sha256(data).hexdigest()})
        return result

    def compare_selection(self, scope, allowed, handles, observed):
        expected = self.describe(scope, allowed, handles)
        if type(observed) is not list or any(type(x) is not dict or set(x) != {'name', 'size'}
                or type(x['name']) is not str or type(x['size']) is not int or x['size'] < 0 for x in observed):
            raise ValueError('OBSERVED_NAME_SIZE_LIST_REQUIRED')
        matched = observed == [{'name': x['name'], 'size': x['size']} for x in expected]
        return {'selection_verified': matched, 'server_saved_verified': False,
                'observation_authenticated': False, 'files': expected,
                'reason': 'names_and_sizes_match' if matched else 'selection_mismatch'}


def compare_download_bytes(data, received, completed, expected_sha256=None):
    """Hash supplied bytes, not a remote URL, path, or invented saved file."""
    data = payload(data)
    if type(received) is not int or received < 0 or type(completed) is not bool:
        raise ValueError('DECLARED_TRANSFER_FIELDS_REQUIRED')
    if expected_sha256 is not None and (type(expected_sha256) is not str or not re.fullmatch('[0-9a-f]{64}', expected_sha256)):
        raise ValueError('LOWERCASE_SHA256_REQUIRED')
    digest = sha256(data).hexdigest()
    reason = ('not_completed' if not completed else 'size_mismatch' if received != len(data)
              else 'digest_mismatch' if expected_sha256 is not None and expected_sha256 != digest
              else 'supplied_bytes_match')
    return {'verified': reason == 'supplied_bytes_match', 'reason': reason, 'size': len(data),
            'sha256': digest, 'expected_digest_checked': expected_sha256 is not None,
            'network_origin_verified': False, 'power_loss_durability_verified': False}