"""Parse and summarize explicitly supplied CronJob generator inputs."""

import base64
import binascii
import configparser
import hashlib
from dataclasses import dataclass
from typing import List, Optional, Sequence, Tuple


class InputInspectionError(ValueError):
    """An input is malformed or cannot be summarized safely."""


@dataclass(frozen=True)
class OpenPgpKeyInfo:
    version: int
    fingerprint: str
    user_ids: Tuple[str, ...]


@dataclass(frozen=True)
class ParsedGpgPublicKey:
    armored: str
    keys: Tuple[OpenPgpKeyInfo, ...]


@dataclass(frozen=True)
class MysqlConnectionInfo:
    host: str
    port: int
    port_is_explicit: bool
    user: str
    database: str


def _crc24(payload: bytes) -> int:
    checksum = 0xB704CE
    for byte in payload:
        checksum ^= byte << 16
        for _ in range(8):
            checksum <<= 1
            if checksum & 0x1000000:
                checksum ^= 0x1864CFB
    return checksum & 0xFFFFFF


def _read_packet_length(payload: bytes, offset: int) -> Tuple[int, int]:
    if offset >= len(payload):
        raise InputInspectionError("OpenPGP packet is missing its length")
    first = payload[offset]
    offset += 1
    if first < 192:
        return first, offset
    if first < 224:
        if offset >= len(payload):
            raise InputInspectionError("OpenPGP packet has a truncated length")
        length = ((first - 192) << 8) + payload[offset] + 192
        return length, offset + 1
    if first == 255:
        if offset + 4 > len(payload):
            raise InputInspectionError("OpenPGP packet has a truncated length")
        return int.from_bytes(payload[offset : offset + 4], "big"), offset + 4
    raise InputInspectionError("partial-length OpenPGP packets are not supported")


def _packets(payload: bytes) -> Tuple[Tuple[int, bytes], ...]:
    result = []
    offset = 0
    while offset < len(payload):
        header = payload[offset]
        offset += 1
        if header & 0x80 == 0:
            raise InputInspectionError("OpenPGP packet has an invalid header")
        if header & 0x40:
            tag = header & 0x3F
            length, offset = _read_packet_length(payload, offset)
        else:
            tag = (header >> 2) & 0x0F
            length_type = header & 0x03
            width = (1, 2, 4)[length_type] if length_type < 3 else None
            if width is None:
                length = len(payload) - offset
            else:
                if offset + width > len(payload):
                    raise InputInspectionError("OpenPGP packet has a truncated length")
                length = int.from_bytes(payload[offset : offset + width], "big")
                offset += width
        end = offset + length
        if end > len(payload):
            raise InputInspectionError("OpenPGP packet body is truncated")
        result.append((tag, payload[offset:end]))
        offset = end
    return tuple(result)


def _fingerprint(public_key_packet: bytes) -> Tuple[int, str]:
    if not public_key_packet:
        raise InputInspectionError("OpenPGP public-key packet is empty")
    version = public_key_packet[0]
    if version == 4:
        if len(public_key_packet) < 6 or len(public_key_packet) > 0xFFFF:
            raise InputInspectionError("OpenPGP v4 public-key packet has an invalid length")
        framed = b"\x99" + len(public_key_packet).to_bytes(2, "big") + public_key_packet
        fingerprint = hashlib.sha1(  # nosec: mandated by the OpenPGP v4 format
            framed,
            usedforsecurity=False,
        ).hexdigest()
    elif version in (5, 6):
        if len(public_key_packet) < 10:
            raise InputInspectionError(
                f"OpenPGP v{version} public-key packet has an invalid length"
            )
        prefix = b"\x9a" if version == 5 else b"\x9b"
        framed = prefix + len(public_key_packet).to_bytes(4, "big") + public_key_packet
        fingerprint = hashlib.sha256(framed).hexdigest()
    else:
        raise InputInspectionError(
            f"unsupported OpenPGP public-key version {version}"
        )
    return version, fingerprint.upper()


def _key_info(packet_stream: Sequence[Tuple[int, bytes]]) -> Tuple[OpenPgpKeyInfo, ...]:
    keys: List[Tuple[int, str, List[str]]] = []
    for tag, body in packet_stream:
        if tag == 6:
            version, fingerprint = _fingerprint(body)
            keys.append((version, fingerprint, []))
        elif tag == 13 and keys:
            try:
                user_id = body.decode("utf-8")
            except UnicodeDecodeError as exc:
                raise InputInspectionError("OpenPGP user ID is not valid UTF-8") from exc
            if not user_id:
                raise InputInspectionError("OpenPGP user ID is empty")
            keys[-1][2].append(user_id)
    if not keys:
        raise InputInspectionError("OpenPGP armor contains no public-key packet")
    return tuple(
        OpenPgpKeyInfo(version, fingerprint, tuple(user_ids))
        for version, fingerprint, user_ids in keys
    )


def parse_gpg_public_key(payload: bytes) -> ParsedGpgPublicKey:
    try:
        text = payload.decode("ascii")
    except UnicodeDecodeError as exc:
        raise InputInspectionError(
            "GPG public key must be an ASCII-armored public key"
        ) from exc
    lines = text.strip().splitlines()
    begin = "-----BEGIN PGP PUBLIC KEY BLOCK-----"
    end = "-----END PGP PUBLIC KEY BLOCK-----"
    if not lines or lines[0] != begin or lines[-1] != end:
        raise InputInspectionError(
            "GPG public key must contain one ASCII-armored public-key block"
        )
    body_lines = lines[1:-1]
    try:
        separator = body_lines.index("")
    except ValueError as exc:
        raise InputInspectionError(
            "GPG public key armor is missing its header separator"
        ) from exc
    for header in body_lines[:separator]:
        if ":" not in header:
            raise InputInspectionError("GPG public key contains an invalid armor header")

    encoded_lines = []
    encoded_checksum: Optional[str] = None
    for line in body_lines[separator + 1 :]:
        if line.startswith("="):
            if encoded_checksum is not None or len(line) != 5:
                raise InputInspectionError(
                    "GPG public key contains an invalid armor checksum"
                )
            encoded_checksum = line[1:]
            continue
        if encoded_checksum is not None or not line:
            raise InputInspectionError("GPG public key contains invalid armored data")
        encoded_lines.append(line)
    if not encoded_lines:
        raise InputInspectionError("GPG public key armor has no packet data")
    try:
        decoded = base64.b64decode("".join(encoded_lines), validate=True)
    except (binascii.Error, ValueError) as exc:
        raise InputInspectionError(
            "GPG public key armor contains invalid base64 data"
        ) from exc
    if encoded_checksum is not None:
        try:
            checksum = base64.b64decode(encoded_checksum, validate=True)
        except (binascii.Error, ValueError) as exc:
            raise InputInspectionError(
                "GPG public key contains an invalid armor checksum"
            ) from exc
        if len(checksum) != 3 or int.from_bytes(checksum, "big") != _crc24(decoded):
            raise InputInspectionError("GPG public key armor checksum does not match")

    keys = _key_info(_packets(decoded))
    return ParsedGpgPublicKey("\n".join(lines) + "\n", keys)


def _unquote_mysql_value(value: Optional[str]) -> str:
    if value is None:
        return ""
    if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
        return value[1:-1]
    return value


def parse_mysql_defaults(payload: bytes) -> MysqlConnectionInfo:
    try:
        text = payload.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise InputInspectionError("my.cnf must be valid UTF-8") from exc
    if "\0" in text:
        raise InputInspectionError("my.cnf contains a NUL byte")
    if any(
        line.lstrip().lower().startswith(("!include", "!includedir"))
        for line in text.splitlines()
    ):
        raise InputInspectionError(
            "my.cnf include directives are not supported in the Pod"
        )
    parser = configparser.RawConfigParser(allow_no_value=True, strict=True)
    parser.optionxform = lambda optionstr: optionstr.lower().replace("_", "-")
    try:
        parser.read_string(text)
    except configparser.Error as exc:
        raise InputInspectionError(
            f"my.cnf is not parseable ({type(exc).__name__}); values were not logged"
        ) from exc
    if not parser.has_section("client"):
        raise InputInspectionError("my.cnf must contain a [client] section")

    values = {}
    for option in ("host", "user", "database"):
        try:
            value = _unquote_mysql_value(parser.get("client", option))
        except (configparser.Error, AttributeError) as exc:
            raise InputInspectionError(
                f"my.cnf [client] must contain a non-empty {option!r} option"
            ) from exc
        if not value or value != value.strip():
            raise InputInspectionError(
                f"my.cnf [client] must contain a non-empty {option!r} option"
            )
        values[option] = value

    port_is_explicit = parser.has_option("client", "port")
    port = 3306
    if port_is_explicit:
        try:
            port = int(_unquote_mysql_value(parser.get("client", "port")))
        except (TypeError, ValueError, configparser.Error) as exc:
            raise InputInspectionError("my.cnf [client] port must be an integer") from exc
        if not 1 <= port <= 65535:
            raise InputInspectionError(
                "my.cnf [client] port must be between 1 and 65535"
            )
    return MysqlConnectionInfo(
        host=values["host"],
        port=port,
        port_is_explicit=port_is_explicit,
        user=values["user"],
        database=values["database"],
    )