#!/usr/bin/env python3
# decode -- tool for decoding various structured binary identifiers
#
# (This should eventually incorporate 'nsap' and 'parse-snmp-engine-id'.)
import argparse
import binascii
import datetime
import enum
import io
import ipaddress
import uuid
from nullroute.io import BinaryReader

# https://www.iana.org/assignments/arp-parameters/arp-parameters.xhtml#arp-parameters-2
class ArpHardwareType(enum.IntEnum):
    NonHardware = 0 # Defined for DHCP client identifiers (RFC 2132)
    Ethernet = 1
    Ieee802 = 6     # *Sometimes* used for Wi-Fi aka IEEE 802.11
    Ieee1394 = 24   # FireWire

# https://www.iana.org/assignments/dhcpv6-parameters/dhcpv6-parameters.xhtml#dhcpv6-parameters-6
class DUIDType(enum.IntEnum):
    LLT = 1         # Link-Layer + Time (RFC 3315)
    EN = 2          # Enterprise Number (RFC 3315)
    LL = 3          # Link Layer (RFC 3315)
    UUID = 4        # Universally Unique ID (RFC 6355)

# https://datatracker.ietf.org/doc/html/rfc4446#section-3.2
class PseudowireType(enum.IntEnum):
    TaggedEthernet  = 0x0004
    Ethernet        = 0x0005

# https://datatracker.ietf.org/doc/html/rfc4446#section-3.3
class PseudowireInterfaceParamSubTlvType(enum.IntEnum):
    InterfaceMTU    = 0x01

# https://datatracker.ietf.org/doc/html/rfc4446#section-3.4.2
class PseudowireAttachmentGroupType(enum.IntEnum):
    RouteDistinguisher  = 0x01

# https://datatracker.ietf.org/doc/html/rfc4446#section-3.4.1
class PseudowireAttachmentIndividualType(enum.IntEnum):
    LocalIdentifier     = 0x01

def try_enum(etype, value):
    try:
        return etype(value)
    except ValueError:
        return value

def show_enum(value, fmt="%d"):
    if hasattr(value, "name"):
        return ("%s (" + fmt + ")") % (value.name, value)
    else:
        return ("%s (" + fmt + ")") % ("unknown", value)

def lookup_iana_pen(vendor):
    return {
        43793: "systemd (Tom Gundersen)",
    }.get(vendor, "unknown")

def parse_hex(s):
    if ":" in s:
        s = s.split(":")
        s = [int(c, 16) for c in s]
        return bytes(s)
    else:
        s = s.replace("-", "")
        s = s.replace(" ", "")
        return binascii.unhexlify(s)

def fmt_hex(buf):
    h = ["%02x" % c for c in buf]
    h = ":".join(h).upper()
    return h or "(empty)"

def is_ascii(buf):
    return all(0x20 < x < 0x7F for x in buf)

def print_hdr(hdr):
    print("\033[1m%s\033[m" % hdr)

def decode_dhcp_client_id(buf):
    Dhcpv4ClientIdDecoder(buf).show()

def decode_dhcpv6_duid(buf):
    Dhcpv6DuidDecoder(buf).show()

def decode_osi_nsap(buf):
    br = BinaryReader(buf)
    afi = br.read_u8()
    if afi == 0x47:
        print("AFI: ISO 6523-ICD IDI (%02x)" % afi)
        icd = br.read_u16_be()
        icd_name = {
            0x0005: "US Federal Government (GOSIP)",
            0x0079: "ATM Forum",
            0x0090: "IANA",
            0x0180: "Lithuania",
        }.get(icd)
        print("IDI (ICD): %s (%04x)" % (icd_name, icd))
        if icd == 0x0005:
            # 47.0005.80FFE1000000F21A26D8.0020EA000EE0.00
            print("HO-DSP (US GOSIP v2):")
            print("  DFI: %s"           % fmt_hex(br.read(1)))
            print("  AdmAuthority: %s"  % fmt_hex(br.read(3)))
            print("  Reserved: %s"      % fmt_hex(br.read(2)))
            print("  RtDomainID: %s"    % fmt_hex(br.read(2)))
            print("  AreaID: %s"        % fmt_hex(br.read(2)))
            print("ID: %s"              % fmt_hex(br.read(6)))
            print("SEL: %s"             % fmt_hex(br.read(1)))
        elif icd == 0x0091:
            print("HO-DSP (Cisco):")
            print("  %s" % fmt_hex(br.read(4)))
            print("  MAC: %s" % fmt_hex(br.read(6)))
            print("ID: %s"              % fmt_hex(br.read(6)))
            print("SEL: %s"             % fmt_hex(br.read(1)))
        else:
            print("Unknown DSP format")
    else:
        print("AFI: unknown (%02x)" % afi)

class Decoder:
    def __init__(self, buf, depth=0):
        self.buf = buf
        self.br = BinaryReader(buf)
        self.depth = depth
        self.hindent = " " * 4 * depth
        self.vindent = " " * 4 * (depth + 1)

    def print_hdr(self, hdr):
        print("%s\033[1m%s\033[m" % (self.hindent, hdr))

    def print_field(self, name, value):
        print("%s%s: %s" % (self.vindent, name, value))

class Dhcpv4ClientIdDecoder(Decoder):
    def show(self):
        # https://tools.ietf.org/html/rfc2132#section-9.14
        # Types (except 0 and 255) correspond to link-layer types, 1 is Ethernet
        self.print_hdr("DHCP Client Identifier")
        id_type = self.br.read_u8()
        id_type = try_enum(ArpHardwareType, id_type)
        id_type_str = {
                0x00: "non-hardware",
                0xFF: "embedded RFC 3315 IAID+DUID",
        }.get(id_type, "unknown")
        if id_type == 0x00:
            self.print_field("Type", "%s (%d)" % (id_type_str, id_type))
            ident = self.br.fh.read()
            if is_ascii(ident):
                self.print_field("Identifier", repr(ident))
            else:
                self.print_field("Identifier", fmt_hex(ident))
        elif id_type == 0xFF:
            # https://tools.ietf.org/html/rfc4361#section-6.1
            self.print_field("Type", "%s (%d)" % (id_type_str, id_type))
            iaid = self.br.read(4)
            self.print_field("IAID", fmt_hex(iaid))
            duid = self.br.fh.read()
            self.print_field("DUID", fmt_hex(duid))
            Dhcpv6DuidDecoder(duid, self.depth+1).show()
        elif id_type == 0x01 and len(buf) == (1 + 16) and buf[1:5] == b'RAS ':
            self.print_field("Type", "Windows RRAS dial-in pool lease")
            magic = self.br.read(4)
            self.print_field("Magic", repr(magic))
            mac = self.br.read(6)
            self.print_field("Server MAC", fmt_hex(mac))
            foo = self.br.read(2)
            self.print_field("Unknown", fmt_hex(foo))
            rest = self.br.fh.read()
            self.print_field("RAS port number", fmt_hex(rest))
        else:
            self.print_field("Type", "link-layer address - %s" % show_enum(id_type))
            mac = self.br.fh.read()
            self.print_field("Link-layer address", fmt_hex(mac))

class Dhcpv6DuidDecoder(Decoder):
    DUID_TYPES = {
        DUIDType.LLT:       "link-layer + time (DUID-LLT)",
        DUIDType.EN:        "vendor-assigned (DUID-EN)",
        DUIDType.LL:        "link-layer (DUID-LL)",
        DUIDType.UUID:      "universally unique (DUID-UUID)",
    }

    def show(self):
        # https://tools.ietf.org/html/rfc3315#section-9.1
        self.print_hdr("DHCPv6 DUID")
        duid_type = self.br.read_u16_be()
        duid_type = try_enum(DUIDType, duid_type)
        duid_type_str = self.DUID_TYPES.get(duid_type, "unknown")
        self.print_field("Type", "%s (0x%04x)" % (duid_type_str, duid_type))
        if duid_type == DUIDType.LLT:
            hwtype = self.br.read_u16_be()
            hwtype = try_enum(ArpHardwareType, hwtype)
            self.print_field("Hardware type", show_enum(hwtype))
            time = self.br.read_u32_be()
            time = datetime.datetime.fromtimestamp(time)
            self.print_field("Time", time)
            lladdr = self.br.fh.read()
            self.print_field("Link-layer address", fmt_hex(lladdr))
        elif duid_type == DUIDType.EN:
            vendor = self.br.read_u32_be()
            self.print_field("Vendor", "%s (%d)" % (lookup_iana_pen(vendor), vendor))
            rest = self.br.fh.read()
            self.print_field("Identifier", fmt_hex(rest))
        elif duid_type == DUIDType.LL:
            hwtype = self.br.read_u16_be()
            hwtype = try_enum(ArpHardwareType, hwtype)
            self.print_field("Hardware type", show_enum(hwtype))
            lladdr = self.br.fh.read()
            self.print_field("Link-layer address", fmt_hex(lladdr))
        elif duid_type == DUIDType.UUID:
            self.buf = br.read(16)
            self.print_field("UUID", uuid.UUID(bytes=buf))
        else:
            self.print_field("Rest", fmt_hex(self.br.fh.read()))

class RouteDistinguisherDecoder(Decoder):
    AFIS = {
        0x0000: "2-byte ASN + 4-byte ID",
        0x0001: "4-byte IPv4 + 2-byte ID",
        0x0002: "4-byte ASN + 2-byte ID",
    }

    def show(self):
        self.print_hdr("Route distinguisher")
        afi = self.br.read_u16_be()
        afi_str = self.AFIS.get(afi, "unknown")
        self.print_field("AFI", "%s (0x%04x)" % (afi_str, afi))
        if afi == 0x0000:
            self.print_field("ASN", "%d" % self.br.read_u16_be())
            self.print_field("ID", "%d" % self.br.read_u32_be())
        elif afi == 0x0001:
            self.print_field("IPv4", ipaddress.IPv4Address(self.br.read(4)))
            self.print_field("ID", self.br.read_u16_be())
        elif afi == 0x0002:
            self.print_field("ASN", self.br.read_u32_be())
            self.print_field("ID", self.br.read_u16_be())
        else:
            self.print_field("Rest", fmt_hex(self.br.fh.read()))

class MplsFecDecoder(Decoder):
    FEC_TYPES = {
        0x80: "RFC 4466 PWid",
        0x81: "RFC 4446 Generalized PWid",
    }

    def show(self):
        self.print_hdr("MPLS FEC")
        fec_type = self.br.read_u8()
        fec_type_str = self.FEC_TYPES.get(fec_type, "unknown")
        self.print_field("Element type", "%s (0x%02x)" % (fec_type_str, fec_type))
        if fec_type == 0x80:
            # https://datatracker.ietf.org/doc/html/rfc4447#section-5.2
            pw_type = self.br.read_u16_be()
            cw_bit = bool(pw_type & 0x8000)
            pw_type = try_enum(PseudowireType, pw_type & ~0x8000)
            self.print_field("Pseudowire type", show_enum(pw_type))
            self.print_field("Control word", cw_bit)
            # Includes PW ID + subTLVs, but not group ID
            info_len = self.br.read_u8()
            self.print_field("PW info length", info_len)
            group_id = self.br.read_u32_be()
            self.print_field("Group ID", group_id)
            if info_len == 0:
                return
            elif info_len < 4:
                self.print_field("Pseudowire ID (too short):", fmt_hex(self.br.fh.read()))
            else:
                pw_id = self.br.read_u32_be()
                self.print_field("Pseudowire ID", pw_id)
                info_len -= 4 # includes pw_id (but not group_id)
                while info_len:
                    assert info_len >= 2, "not enough bytes for TLV type+len"
                    tlv_type = self.br.read_u8()
                    tlv_type = try_enum(PseudowireInterfaceParamSubTlvType, tlv_type)
                    tlv_len = self.br.read_u8()
                    assert tlv_len >= 2, "TLV impossibly short"
                    assert tlv_len <= info_len, "not enough bytes left for TLV data"
                    tlv_data = self.br.read(tlv_len - 2)
                    self.print_field("TLV", "%s = %s" % (show_enum(tlv_type, "0x%02d"),
                                                         fmt_hex(tlv_data)))
                    info_len -= tlv_len
        elif fec_type == 0x81:
            # https://datatracker.ietf.org/doc/html/rfc4447#section-5.3
            pw_type = self.br.read_u16_be()
            cw_bit = bool(pw_type & 0x8000)
            pw_type = try_enum(PseudowireType, pw_type & ~0x8000)
            self.print_field("Pseudowire type", show_enum(pw_type))
            self.print_field("Control word", cw_bit)
            # Includes AGI + SAII + TAII
            info_len = self.br.read_u8()
            self.print_field("PW info length", info_len)
            if info_len == 0:
                return
            else:
                assert info_len >= 2, "not enough bytes for AGI type+len"
                agi_type = self.br.read_u8()
                agi_type = try_enum(PseudowireAttachmentGroupType, agi_type)
                agi_len = self.br.read_u8()
                info_len -= 2
                assert agi_len <= info_len, "not enough bytes left for AGI value"
                agi_value = self.br.read(agi_len)
                info_len -= agi_len
                self.print_field("AGI type", show_enum(agi_type))
                if agi_type == 0x01:
                    RouteDistinguisherDecoder(agi_value, self.depth+1).show()
                else:
                    self.print_field("Raw AGI value", fmt_hex(agi_value))

                for t in ["Source", "Dest"]:
                    assert info_len >= 2, "not enough bytes for AII type+len"
                    aii_type = self.br.read_u8()
                    aii_type = try_enum(PseudowireAttachmentIndividualType, aii_type)
                    aii_len = self.br.read_u8()
                    info_len -= 2
                    assert aii_len <= info_len, "not enough bytes left for AII value"
                    self.print_field("%s AII type" % t, show_enum(aii_type))
                    if aii_type == 0x01 and aii_len == 4:
                        aii_value = self.br.read_u32_be()
                        info_len -= aii_len
                        self.print_field("%s AII" % t, aii_value)
                    else:
                        aii_value = self.br.read(aii_len)
                        info_len -= aii_len
                        self.print_field("%s AII" % t, fmt_hex(aii_value))
        else:
            self.print_field("Rest", fmt_hex(self.br.fh.read()))

parser = argparse.ArgumentParser()
parser.add_argument("--dhcp4", dest="type", action="store_const", const="dhcp-client-id")
parser.add_argument("--dhcp6", dest="type", action="store_const", const="dhcpv6-duid")
parser.add_argument("--bgp-rd", dest="type", action="store_const", const="bgp-rd")
parser.add_argument("--mpls-fec", dest="type", action="store_const", const="mpls-fec")
parser.add_argument("data")
args = parser.parse_args()
data = parse_hex(args.data.replace(".", ""))

if args.type == "dhcp-client-id":
    decode_dhcp_client_id(data)
elif args.type == "dhcpv6-duid":
    decode_dhcpv6_duid(data)
elif args.type == "osi-nsap":
    decode_osi_nsap(data)
elif args.type == "bgp-rd":
    RouteDistinguisherDecoder(data).show()
elif args.type == "mpls-fec":
    MplsFecDecoder(data).show()
else:
    exit("error: Unknown type %r" % args.type)
