trestle.core.signing

def build_in_toto_statement(
    subject_name: str, digest: str, digest_algorithm: Algorithm = DEFAULT_DIGEST_ALGORITHM
) -> Dict[str, Any]:
    """Build the in-toto Statement payload for a JSON artifact digest."""
    algorithm_name = digest_algorithm_name(digest_algorithm)
    return {
        '_type': IN_TOTO_STATEMENT_TYPE,
        'subject': [{'name': subject_name, 'digest': {algorithm_name: digest}}],
        'predicateType': OSCAL_PREDICATE_TYPE,
        'predicate': {
            'canonicalization': CANONICALIZATION_ALGORITHM,
            'digestAlgorithm': algorithm_name,
            'digestSource': DIGEST_SOURCE,
            'tool': 'compliance-trestle',
        },
    }

def create_oscal_provenance_envelope(
    input_path: pathlib.Path,
    signer: Signer,
    subject_name: Optional[str] = None,
    digest_algorithm: Algorithm = DEFAULT_DIGEST_ALGORITHM,
) -> Dict[str, Any]:
    """Create a DSSE envelope whose payload is an in-toto Statement for a JSON file."""
    _validate_json_path(input_path)
    _, canonical_bytes = load_canonical_json_file(input_path)
    digest = digest_hex(canonical_bytes, digest_algorithm)
    statement = build_in_toto_statement(subject_name or input_path.name, digest, digest_algorithm)
    return sign_in_toto_statement(statement, signer)

def dsse_pae(payload_type: str, payload: bytes) -> bytes:
    """Return DSSE pre-authentication encoding for a payload type and payload."""
    payload_type_bytes = payload_type.encode(const.FILE_ENCODING)
    return b' '.join(
        [
            b'DSSEv1',
            str(len(payload_type_bytes)).encode(const.FILE_ENCODING),
            payload_type_bytes,
            str(len(payload)).encode(const.FILE_ENCODING),
            payload,
        ]
    )

def load_dsse_envelope(envelope_path: pathlib.Path) -> Dict[str, Any]:
    """Load a DSSE envelope from JSON."""
    envelope, _ = canonicalize_json_text(envelope_path.read_text(encoding=const.FILE_ENCODING))

    if not isinstance(envelope, dict):
        raise TrestleError('DSSE envelope must be a JSON object.')
    return envelope

def load_in_toto_statement(payload: bytes) -> Dict[str, Any]:
    """Load a verified DSSE payload as an in-toto Statement object."""
    return _load_statement(payload)

def load_pem_private_key_signer(path: pathlib.Path, password: Optional[str] = None) -> Signer:
    """Load a local PEM private key as a securesystemslib signer."""
    password_bytes = password.encode(const.FILE_ENCODING) if password is not None else None
    try:
        private_key = serialization.load_pem_private_key(path.read_bytes(), password=password_bytes)
        return CryptoSigner(private_key)
    except Exception as error:
        raise TrestleError(f'Unable to load private key for signing: {path}') from error

def load_pem_public_key(path: pathlib.Path) -> Key:
    """Load a local PEM public key as a securesystemslib key."""
    try:
        public_key = serialization.load_pem_public_key(path.read_bytes())
        return SSlibKey.from_crypto(public_key)
    except Exception as error:
        raise TrestleError(f'Unable to load public key for verification: {path}') from error

def sign_in_toto_statement(statement: Dict[str, Any], signer: Signer) -> Dict[str, Any]:
    """Sign an in-toto Statement as a DSSE envelope."""
    payload = _json_bytes(statement)
    signature = signer.sign(dsse_pae(DSSE_PAYLOAD_TYPE, payload))
    return {
        'payload': _b64_encode(payload),
        'payloadType': DSSE_PAYLOAD_TYPE,
        'signatures': [_dsse_signature(signature)],
    }

def verify_dsse_payload(envelope: Dict[str, Any], public_key: Key) -> bytes:
    """Verify a DSSE envelope signature and return its payload bytes."""
    return _verified_payload(envelope, public_key)

def verify_oscal_provenance_envelope(
    input_path: pathlib.Path,
    envelope: Dict[str, Any],
    public_key: Key,
    subject_name: Optional[str] = None,
    digest_algorithm: Optional[Algorithm] = None,
) -> Dict[str, Any]:
    """Verify a JSON file, detecting the signed digest algorithm unless one is required."""
    _validate_json_path(input_path)
    payload = _verified_payload(envelope, public_key)
    statement = _load_statement(payload)
    _validate_statement(statement, subject_name or input_path.name, input_path, digest_algorithm)
    return statement

def write_dsse_envelope(envelope: Dict[str, Any], output_path: pathlib.Path, overwrite: bool = False) -> None:
    """Write a DSSE envelope as JSON, optionally replacing an existing file."""
    if output_path.is_symlink():
        raise TrestleError(f'DSSE output path must not be a symlink: {output_path}')
    if output_path.exists():
        if output_path.is_dir():
            raise TrestleError(f'DSSE output path is a directory: {output_path}')
        if not overwrite:
            raise TrestleError(f'DSSE output path already exists: {output_path}')
    output_path.parent.mkdir(parents=True, exist_ok=True)
    serialized_envelope = json.dumps(envelope, indent=2, sort_keys=True) + '\n'
    if overwrite:
        temporary_path: Optional[pathlib.Path] = None
        try:
            with tempfile.NamedTemporaryFile(
                'w', dir=output_path.parent, delete=False, encoding=const.FILE_ENCODING
            ) as temporary_file:
                temporary_path = pathlib.Path(temporary_file.name)
                temporary_file.write(serialized_envelope)
            temporary_path.replace(output_path)
        finally:
            if temporary_path and temporary_path.exists():
                temporary_path.unlink()
    else:
        try:
            with output_path.open('x', encoding=const.FILE_ENCODING) as write_file:
                write_file.write(serialized_envelope)
        except FileExistsError as error:
            raise TrestleError(f'DSSE output path already exists: {output_path}') from error