# SPDX-License-Identifier: GPL-2.0-only
# This file is part of Scapy
# See https://scapy.net/ for more information
# Copyright (C) Gabriel Potter <gabriel[]potter[]fr>

"""
Generic Security Services (GSS) API

Implements parts of:

    - GSSAPI: RFC4121 / RFC2743
    - GSSAPI C bindings: RFC2744
    - Channel Bindings for TLS: RFC5929

This is implemented in the following SSPs:

    - :class:`~scapy.layers.ntlm.NTLMSSP`
    - :class:`~scapy.layers.kerberos.KerberosSSP`
    - :class:`~scapy.layers.spnego.SPNEGOSSP`
    - :class:`~scapy.layers.msrpce.msnrpc.NetlogonSSP`

.. note::
    You will find more complete documentation for this layer over at
    `GSSAPI <https://scapy.readthedocs.io/en/latest/layers/gssapi.html>`_
"""

import abc

from dataclasses import dataclass
from enum import Enum, IntEnum, IntFlag

from scapy.asn1.asn1 import (
    ASN1_SEQUENCE,
    ASN1_Class_UNIVERSAL,
    ASN1_Codecs,
)
from scapy.asn1.ber import BERcodec_SEQUENCE, BER_id_dec
from scapy.asn1.mib import conf  # loads conf.mib
from scapy.asn1fields import (
    ASN1F_OID,
    ASN1F_PACKET,
    ASN1F_SEQUENCE,
)
from scapy.asn1packet import ASN1_Packet
from scapy.error import log_runtime
from scapy.fields import (
    FieldLenField,
    LEIntEnumField,
    PacketField,
    StrLenField,
)
from scapy.packet import Packet

# Type hints
from typing import (
    Any,
    List,
    Optional,
    Tuple,
)

# https://datatracker.ietf.org/doc/html/rfc1508#page-48


class ASN1_Class_GSSAPI(ASN1_Class_UNIVERSAL):
    name = "GSSAPI"
    APPLICATION = 0x60


class ASN1_GSSAPI_APPLICATION(ASN1_SEQUENCE):
    tag = ASN1_Class_GSSAPI.APPLICATION


class BERcodec_GSSAPI_APPLICATION(BERcodec_SEQUENCE):
    tag = ASN1_Class_GSSAPI.APPLICATION


class ASN1F_GSSAPI_APPLICATION(ASN1F_SEQUENCE):
    ASN1_tag = ASN1_Class_GSSAPI.APPLICATION


# GSS API Blob
# https://datatracker.ietf.org/doc/html/rfc4121

# Filled by providers
_GSSAPI_OIDS = {}
_GSSAPI_SIGNATURE_OIDS = {}

# section 4.1


class GSSAPI_BLOB(ASN1_Packet):
    ASN1_codec = ASN1_Codecs.BER
    ASN1_root = ASN1F_GSSAPI_APPLICATION(
        ASN1F_OID("MechType", "1.3.6.1.5.5.2"),
        ASN1F_PACKET(
            "innerToken",
            None,
            None,
            next_cls_cb=lambda pkt: _GSSAPI_OIDS.get(pkt.MechType.val, conf.raw_layer),
        ),
    )

    @classmethod
    def dispatch_hook(cls, _pkt=None, *args, **kargs):
        if _pkt and len(_pkt) >= 1:
            if _pkt[0] & 0xA0 >= 0xA0:
                from scapy.layers.spnego import SPNEGO_negToken

                # XXX: sometimes the token is raw, we should look from
                # the session what to use here. For now: hardcode SPNEGO
                # (THIS IS A VERY STRONG ASSUMPTION)
                return SPNEGO_negToken
            elif _pkt[:7] == b"NTLMSSP":
                from scapy.layers.ntlm import NTLM_Header

                # XXX: if no mechTypes are provided during SPNEGO exchange,
                # Windows falls back to a plain NTLM_Header.
                return NTLM_Header.dispatch_hook(_pkt=_pkt, *args, **kargs)
            elif BER_id_dec(_pkt)[0] & 0x7F > 0x60:
                from scapy.layers.kerberos import Kerberos

                # XXX: Heuristic to detect raw Kerberos packets, when Windows
                # fallsback or when the parent data hasn't got any mechtype specified.
                return Kerberos
        return cls


# Same but to store the signatures (e.g. DCE/RPC)


class GSSAPI_BLOB_SIGNATURE(ASN1_Packet):
    ASN1_codec = ASN1_Codecs.BER
    ASN1_root = ASN1F_GSSAPI_APPLICATION(
        ASN1F_OID("MechType", "1.3.6.1.5.5.2"),
        ASN1F_PACKET(
            "innerToken",
            None,
            None,
            next_cls_cb=lambda pkt: _GSSAPI_SIGNATURE_OIDS.get(
                pkt.MechType.val, conf.raw_layer
            ),  # noqa: E501
        ),
    )

    @classmethod
    def dispatch_hook(cls, _pkt=None, *args, **kargs):
        if _pkt and len(_pkt) >= 2:
            # Sometimes the token is raw. Detect that with educated
            # heuristics.
            if _pkt[:2] in [b"\x04\x04", b"\x05\x04"]:
                from scapy.layers.kerberos import KRB_InnerToken

                return KRB_InnerToken
            elif len(_pkt) >= 4 and _pkt[:4] == b"\x01\x00\x00\x00":
                from scapy.layers.ntlm import NTLMSSP_MESSAGE_SIGNATURE

                return NTLMSSP_MESSAGE_SIGNATURE
        return cls


class _GSSAPI_Field(PacketField):
    """
    PacketField that contains a GSSAPI_BLOB_SIGNATURE, but one that can
    have a payload when not encrypted.
    """

    __slots__ = ["pay_cls"]

    def __init__(self, name, pay_cls):
        self.pay_cls = pay_cls
        super().__init__(
            name,
            None,
            GSSAPI_BLOB_SIGNATURE,
        )

    def getfield(self, pkt, s):
        remain, val = super().getfield(pkt, s)
        if remain and val:
            val.payload = self.pay_cls(remain)
            return b"", val
        return remain, val


# RFC2744 Annex A, Null values

GSS_C_QOP_DEFAULT = 0
GSS_C_NO_CHANNEL_BINDINGS = b"\x00"


# RFC2744 sect 3.9 - Status Values

GSS_S_COMPLETE = 0

# These errors are encoded into the 32-bit GSS status code as follows:
#   MSB                                                        LSB
#   |------------------------------------------------------------|
#   |  Calling Error | Routine Error  |    Supplementary Info    |
#   |------------------------------------------------------------|
# Bit 31            24 23            16 15                       0

GSS_C_CALLING_ERROR_OFFSET = 24
GSS_C_ROUTINE_ERROR_OFFSET = 16
GSS_C_SUPPLEMENTARY_OFFSET = 0

# Calling errors:

GSS_S_CALL_INACCESSIBLE_READ = 1 << GSS_C_CALLING_ERROR_OFFSET
GSS_S_CALL_INACCESSIBLE_WRITE = 2 << GSS_C_CALLING_ERROR_OFFSET
GSS_S_CALL_BAD_STRUCTURE = 3 << GSS_C_CALLING_ERROR_OFFSET

# Routine errors:

GSS_S_BAD_MECH = 1 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_BAD_NAME = 2 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_BAD_NAMETYPE = 3 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_BAD_BINDINGS = 4 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_BAD_STATUS = 5 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_BAD_SIG = 6 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_BAD_MIC = GSS_S_BAD_SIG
GSS_S_NO_CRED = 7 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_NO_CONTEXT = 8 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_DEFECTIVE_TOKEN = 9 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_DEFECTIVE_CREDENTIAL = 10 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_CREDENTIALS_EXPIRED = 11 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_CONTEXT_EXPIRED = 12 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_FAILURE = 13 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_BAD_QOP = 14 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_UNAUTHORIZED = 15 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_UNAVAILABLE = 16 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_DUPLICATE_ELEMENT = 17 << GSS_C_ROUTINE_ERROR_OFFSET
GSS_S_NAME_NOT_MN = 18 << GSS_C_ROUTINE_ERROR_OFFSET

# Supplementary info bits:

GSS_S_CONTINUE_NEEDED = 1 << (GSS_C_SUPPLEMENTARY_OFFSET + 0)
GSS_S_DUPLICATE_TOKEN = 1 << (GSS_C_SUPPLEMENTARY_OFFSET + 1)
GSS_S_OLD_TOKEN = 1 << (GSS_C_SUPPLEMENTARY_OFFSET + 2)
GSS_S_UNSEQ_TOKEN = 1 << (GSS_C_SUPPLEMENTARY_OFFSET + 3)
GSS_S_GAP_TOKEN = 1 << (GSS_C_SUPPLEMENTARY_OFFSET + 4)

# Address families (RFC2744 sect 3.11)

_GSS_ADDRTYPE = {
    0: "GSS_C_AF_UNSPEC",
    1: "GSS_C_AF_LOCAL",
    2: "GSS_C_AF_INET",
    3: "GSS_C_AF_IMPLINK",
    4: "GSS_C_AF_PUP",
    5: "GSS_C_AF_CHAOS",
    6: "GSS_C_AF_NS",
    7: "GSS_C_AF_NBS",
    8: "GSS_C_AF_ECMA",
    9: "GSS_C_AF_DATAKIT",
    10: "GSS_C_AF_CCITT",
    11: "GSS_C_AF_SNA",
    12: "GSS_C_AF_DECnet",
    13: "GSS_C_AF_DLI",
    14: "GSS_C_AF_LAT",
    15: "GSS_C_AF_HYLINK",
    16: "GSS_C_AF_APPLETALK",
    17: "GSS_C_AF_BSC",
    18: "GSS_C_AF_DSS",
    19: "GSS_C_AF_OSI",
    21: "GSS_C_AF_X25",
    255: "GSS_C_AF_NULLADDR",
}


# GSS Structures


class ChannelBindingType(Enum):
    """
    Channel Binding Application Data types, per:
    RFC 5929 / RFC 9266
    """

    TLS_UNIQUE = "unique"
    TLS_SERVER_END_POINT = "tls-server-end-point"
    TLS_UNIQUE_FOR_TELNET = "tls-unique-for-telnet"
    TLS_EXPORTER = "tls-exporter"  # RFC9266


class GssBufferDesc(Packet):
    name = "gss_buffer_desc"
    fields_desc = [
        FieldLenField("length", None, length_of="value", fmt="<I"),
        StrLenField("value", "", length_from=lambda pkt: pkt.length),
    ]

    def default_payload_class(self, payload):
        return conf.padding_layer


class GssChannelBindings(Packet):
    name = "gss_channel_bindings_struct"
    fields_desc = [
        LEIntEnumField("initiator_addrtype", 0, _GSS_ADDRTYPE),
        PacketField("initiator_address", GssBufferDesc(), GssBufferDesc),
        LEIntEnumField("acceptor_addrtype", 0, _GSS_ADDRTYPE),
        PacketField("acceptor_address", GssBufferDesc(), GssBufferDesc),
        PacketField("application_data", None, GssBufferDesc),
    ]

    def digestMD5(self):
        """
        Calculate a MD5 hash of the channel binding
        """
        from scapy.layers.tls.crypto.hash import Hash_MD5

        return Hash_MD5().digest(bytes(self))

    @classmethod
    def fromssl(
        cls,
        token_type: ChannelBindingType,
        sslsock=None,
        certfile=None,
    ) -> "GssChannelBindings":
        """
        Build a GssChannelBindings struct from a socket

        :param token_type: the type from ChannelBindingType, per RFC5929
        :param sslsock: take the certificate from the the socket.socket object
        :param certfile: take the certificate from a file
        """
        from scapy.layers.tls.cert import Cert
        from cryptography.hazmat.primitives import hashes

        if token_type == ChannelBindingType.TLS_SERVER_END_POINT:
            # RFC5929 sect 4
            try:
                # Parse certificate
                if certfile is not None:
                    cert = Cert(certfile)
                else:
                    cert = Cert(sslsock.getpeercert(binary_form=True))
            except Exception:
                # We failed to parse the certificate.
                log_runtime.warning("Failed to parse the SSL Certificate. CBT not used")
                return GSS_C_NO_CHANNEL_BINDINGS
            try:
                h = cert.getSignatureHash()
            except Exception:
                # We failed to get the signature algorithm.
                log_runtime.warning(
                    "Failed to get the Certificate signature algorithm. CBT not used"
                )
                return GSS_C_NO_CHANNEL_BINDINGS
            # RFC5929 sect 4.1
            if h == hashes.MD5 or h == hashes.SHA1:
                h = hashes.SHA256
            # Get bytes of first certificate if there are multiple
            c = cert.x509Cert.copy()
            c.remove_payload()
            cdata = bytes(c)
            # Calc hash of certificate
            digest = hashes.Hash(h)
            digest.update(cdata)
            cbdata = digest.finalize()
        elif token_type == ChannelBindingType.TLS_UNIQUE:
            # RFC5929 sect 3
            cbdata = sslsock.get_channel_binding(cb_type="tls-unique")
        else:
            raise NotImplementedError
        # RFC5056 sect 2.1
        # "channel bindings MUST start with the channel binding unique prefix followed
        # by a colon (ASCII 0x3A)."
        return GssChannelBindings(
            application_data=GssBufferDesc(
                value=token_type.value.encode() + b":" + cbdata
            )
        )


# --- The base GSSAPI SSP base class


class GSS_C_FLAGS(IntFlag):
    """
    Authenticator Flags per RFC2744 req_flags
    """

    GSS_C_DELEG_FLAG = 0x01
    GSS_C_MUTUAL_FLAG = 0x02
    GSS_C_REPLAY_FLAG = 0x04
    GSS_C_SEQUENCE_FLAG = 0x08
    GSS_C_CONF_FLAG = 0x10  # confidentiality
    GSS_C_INTEG_FLAG = 0x20  # integrity
    # RFC4757
    GSS_C_DCE_STYLE = 0x1000
    GSS_C_IDENTIFY_FLAG = 0x2000
    GSS_C_EXTENDED_ERROR_FLAG = 0x4000


class GSS_S_FLAGS(IntFlag):
    """
    Equivalent to Microsoft's ASC_REQ* Flags in AcceptSecurityContext
    """

    GSS_S_ALLOW_MISSING_BINDINGS = 0x10000000


class SSP:
    """
    The general SSP class
    """

    auth_type = 0x00

    def __init__(self, **kwargs):
        if kwargs:
            raise ValueError("Unknown SSP parameters: " + ",".join(list(kwargs)))

    def __repr__(self):
        return "<%s>" % self.__class__.__name__

    class CONTEXT:
        """
        A Security context i.e. the 'state' of the secure negotiation
        """

        __slots__ = ["state", "_flags", "passive"]

        def __init__(self, req_flags: Optional["GSS_C_FLAGS | GSS_S_FLAGS"] = None):
            if req_flags is None:
                # Default
                req_flags = (
                    GSS_C_FLAGS.GSS_C_EXTENDED_ERROR_FLAG
                    | GSS_C_FLAGS.GSS_C_MUTUAL_FLAG
                )
            self.flags = req_flags
            self.passive = False

        def clifailure(self):
            # This allows to reset the client context without discarding it.
            pass

        # 'flags' is the most important attribute. Use a setter to sanitize it.

        @property
        def flags(self):
            return self._flags

        @flags.setter
        def flags(self, x):
            self._flags = GSS_C_FLAGS(int(x))

        def __repr__(self):
            return "[Default SSP]"

    class STATE(IntEnum):
        """
        An Enum that contains the states of an SSP
        """

    @abc.abstractmethod
    def GSS_Init_sec_context(
        self,
        Context: CONTEXT,
        input_token=None,
        target_name: Optional[str] = None,
        req_flags: Optional[GSS_C_FLAGS] = None,
        chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,
    ):
        """
        GSS_Init_sec_context: client-side call for the SSP
        """
        raise NotImplementedError

    @abc.abstractmethod
    def GSS_Accept_sec_context(
        self,
        Context: CONTEXT,
        input_token=None,
        req_flags: Optional[GSS_S_FLAGS] = GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS,
        chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,
    ):
        """
        GSS_Accept_sec_context: server-side call for the SSP
        """
        raise NotImplementedError

    @abc.abstractmethod
    def GSS_Inquire_names_for_mech(self) -> List[str]:
        """
        Get the available OIDs for this mech, in order of preference.
        """
        raise NotImplementedError

    # Passive

    @abc.abstractmethod
    def GSS_Passive(
        self,
        Context: CONTEXT,
        input_token=None,
    ):
        """
        GSS_Passive: client/server call for the SSP in passive mode
        """
        raise NotImplementedError

    def GSS_Passive_set_Direction(self, Context: CONTEXT, IsAcceptor=False):
        """
        GSS_Passive_set_Direction: used to swap the direction in passive mode
        """
        pass

    # MS additions (*Ex functions)

    @dataclass
    class WRAP_MSG:
        conf_req_flag: bool
        sign: bool
        data: bytes

    @abc.abstractmethod
    def GSS_WrapEx(
        self,
        Context: CONTEXT,
        msgs: List[WRAP_MSG],
        qop_req: int = GSS_C_QOP_DEFAULT,
    ) -> Tuple[List[WRAP_MSG], Any]:
        """
        GSS_WrapEx

        :param Context: the SSP context
        :param qop_req: int (0 specifies default QOP)
        :param msgs: list of WRAP_MSG

        :returns: (data, signature)
        """
        raise NotImplementedError

    @abc.abstractmethod
    def GSS_UnwrapEx(
        self, Context: CONTEXT, msgs: List[WRAP_MSG], signature
    ) -> List[WRAP_MSG]:
        """
        :param Context: the SSP context
        :param msgs: list of WRAP_MSG
        :param signature: the signature

        :raises ValueError: if MIC failure.
        :returns: data
        """
        raise NotImplementedError

    @dataclass
    class MIC_MSG:
        sign: bool
        data: bytes

    @abc.abstractmethod
    def GSS_GetMICEx(
        self,
        Context: CONTEXT,
        msgs: List[MIC_MSG],
        qop_req: int = GSS_C_QOP_DEFAULT,
    ) -> Any:
        """
        GSS_GetMICEx

        :param Context: the SSP context
        :param qop_req: int (0 specifies default QOP)
        :param msgs: list of VERIF_MSG

        :returns: signature
        """
        raise NotImplementedError

    @abc.abstractmethod
    def GSS_VerifyMICEx(
        self,
        Context: CONTEXT,
        msgs: List[MIC_MSG],
        signature,
    ) -> None:
        """
        :param Context: the SSP context
        :param msgs: list of VERIF_MSG
        :param signature: the signature

        :raises ValueError: if MIC failure.
        """
        raise NotImplementedError

    @abc.abstractmethod
    def MaximumSignatureLength(self, Context: CONTEXT):
        """
        Returns the Maximum Signature length.

        This will be used in auth_len in DceRpc5, and is necessary for
        PFC_SUPPORT_HEADER_SIGN to work properly.
        """
        raise NotImplementedError

    # RFC 2743

    # sect 2.3.1

    def GSS_GetMIC(
        self,
        Context: CONTEXT,
        message: bytes,
        qop_req: int = GSS_C_QOP_DEFAULT,
    ):
        """
        See GSS_GetMICEx
        """
        return self.GSS_GetMICEx(
            Context,
            [
                self.MIC_MSG(
                    sign=True,
                    data=message,
                )
            ],
            qop_req=qop_req,
        )

    # sect 2.3.2

    def GSS_VerifyMIC(
        self,
        Context: CONTEXT,
        message: bytes,
        signature,
    ) -> None:
        """
        See GSS_VerifyMICEx
        """
        self.GSS_VerifyMICEx(
            Context,
            [
                self.MIC_MSG(
                    sign=True,
                    data=message,
                )
            ],
            signature,
        )

    # sect 2.3.3

    def GSS_Wrap(
        self,
        Context: CONTEXT,
        input_message: bytes,
        conf_req_flag: bool,
        qop_req: int = GSS_C_QOP_DEFAULT,
    ):
        """
        See GSS_WrapEx
        """
        _msgs, signature = self.GSS_WrapEx(
            Context,
            [
                self.WRAP_MSG(
                    conf_req_flag=conf_req_flag,
                    sign=True,
                    data=input_message,
                )
            ],
            qop_req=qop_req,
        )
        if _msgs[0].data:
            signature /= _msgs[0].data
        return signature

    # sect 2.3.4

    def GSS_Unwrap(
        self,
        Context: CONTEXT,
        signature,
    ):
        """
        See GSS_UnwrapEx
        """
        data = b""
        if signature.payload:
            # signature has a payload that is the data. Let's get that payload
            # in its original form, and use it for verifying the checksum.
            if signature.payload.original:
                data = signature.payload.original
            else:
                data = bytes(signature.payload)
            signature = signature.copy()
            signature.remove_payload()
        return self.GSS_UnwrapEx(
            Context,
            [
                self.WRAP_MSG(
                    conf_req_flag=True,
                    sign=True,
                    data=data,
                )
            ],
            signature,
        )[0].data

    # MISC

    def NegTokenInit2(self):
        """
        Server-Initiation
        See [MS-SPNG] sect 3.2.5.2
        """
        return None, None

    def SupportsMechListMIC(self):
        """
        Returns whether mechListMIC is supported or not
        """
        return True

    def GetMechListMIC(self, Context, input):
        """
        Compute mechListMIC
        """
        return self.GSS_GetMIC(Context, input)

    def VerifyMechListMIC(self, Context, otherMIC, input):
        """
        Verify mechListMIC
        """
        return self.GSS_VerifyMIC(Context, input, otherMIC)

    def LegsAmount(self, Context: CONTEXT):
        """
        Returns the amount of 'legs' (how MS calls it) of the SSP.

        i.e. 2 for Kerberos, 3 for NTLM and Netlogon
        """
        return 2