# Copyright 2026 SigmaSRC, Inc. All Rights Reserved.

"""Converters and validators for standard parameter types."""

import sys

from sigma.util.fuelset import FuelSet
from sigma.runtime.pwall import PoliWall, RangeSet, IPAddr
from sigma.runtime.pwall import ipaddr, ipaddrs, parse_range

class ParamError(Exception):
    """Base class for exceptions raised here."""

class BadTypeName(ParamError):
    """Exception raised if the type name is invalid."""

class BadValue(ParamError):
    """Exception raised if the value is invalid."""

def _convert_int(value):
    import re
    stripped = value.strip()
    if re.match(r'^-?0\d+$', stripped):
        sign = '-' if stripped.startswith('-') else ''
        digits = stripped.lstrip('-').lstrip('0')
        value = sign + '0o' + (digits or '0')
    return int(value, 0)

def _convert_str(value):
    return str(value)

def _convert_float(value):
    return float(value)

_boolmap = {"false": False,
            "true": True,
            "f": False,
            "t": True,
            "off": False,
            "on": True,
            "no": False,
            "yes": True,
            "n": False,
            "y": True,
            "0": False,
            "1": True}

def _convert_bool(value):
    v = value.strip().lower()
    if v not in _boolmap:
        raise ValueError("Invalid bool value: %s" % value)
    return _boolmap[v]

def _convert_IPAddr(value):
    return ipaddr(value)

def _strip_curlies(value):
    value = value.strip()
    if value[:1] + value[-1:] != "{}":
        raise ValueError("set needs to be enclosed in {}")
    return value[1:-1]

def _convert_set_of_IPAddr(value):
    value = value.strip()
    if not value.startswith("{"):
        return _get_hostgroup_by_name(value)
    value = _strip_curlies(value)
    result = RangeSet()
    for part in str(value).split(","):
        if not part.strip():
            continue
        if "-" in part:
            lo, hi = part.split("-", 1)
            lo = ipaddr(lo.strip(), True)
            hi = ipaddr(hi.strip(), True)
        else:
            lo, hi = parse_range(part.strip())
            lo = ipaddr(lo)
            hi = ipaddr(hi)
        result.add(lo, hi)
    return result

def _convert_set_of_int(value):
    # XXX This doesn't support negative integers; Dan doesn't need them (yet)
    value = _strip_curlies(value)
    result = RangeSet()
    for part in str(value).split(","):
        if not part.strip():
            continue
        if "-" in part:
            lo, hi = part.split("-", 1)
            lo = int(lo.strip())
            hi = int(hi.strip())
        else:
            lo = int(part.strip())
            hi = lo
        result.add(lo, hi)
    return result

def _convert_set_of_str(value):
    # This should accept exactly the syntax used for 'set of str' in Fuel
    value = _strip_curlies(value)
    try:
        # Believe me, this evaluation is safe
        items = eval("[" + value + "]", {"__builtins__": {}})
    except KeyboardInterrupt:
        raise
    except Exception as err:
        raise ValueError(str(err))
    for x in items:
        if not isinstance(x, str):
            raise ValueError("Invalid string literal %r" % x)
    return FuelSet(items)

_matrix = None

def _get_hostgroup_by_name(name):
    if name == "MY_LAN":
        return PoliWall.MY_LAN
    global _matrix
    if _matrix is None:
        try:
            from sigma.runtime.matrix import loadmatrix
            _matrix = loadmatrix()
        except Exception as err:
            # If we can't access the matrix, use a fake object.
            # The most likely reason for this is that there is no
            # config file or no Sigma root directory.
            print("Can't acces matrix:", err, file=sys.stderr)
            class FakeMatrix(object):
                def getid(self): return 1 # See below
            _matrix = FakeMatrix()
    if _matrix.getid() == 1:
        # If the matrix has never been initialized, return an empty
        # group, thereby bypassing the group name validation; that's
        # better than refusing *all* group names...
        return FuelSet()
    group = _matrix.gethostgroup(name)
    if group is None:
        raise ValueError("Unknown host group name %r" % name)
    return group

def _check_enum(typename, value, mapping):
    if value in mapping:
        return mapping[value]
    raise ValueError("Invalid %s value %r; valid values are %s" %
                     (typename, value, mapping.keys()))

_map_Action = {
    "allow": FuelSet(["A"]),         # pw.ALLOW
    "deny":  FuelSet(["D", "B"]),    # pw.DENY | pw.LOG
    "deny_silently": FuelSet(["D"]), # pw.DENY
    }

def _convert_Action(value):
    return _check_enum("Action", value, _map_Action)

_alias_Action = "set of str"

_map_Direction = {
    "to":   FuelSet(["to"]),
    "from": FuelSet(["from"]),
    "both": FuelSet(["from", "to"]),
    }

def _convert_Direction(value):
    return _check_enum("Direction", value, _map_Direction)

_alias_Direction = "set of str"

_map_PortProtocols = {
    "tcp":  FuelSet(["tcp"]),
    "udp":  FuelSet(["udp"]),
    "both": FuelSet(["tcp", "udp"]),
    }

def _convert_PortProtocols(value):
    return _check_enum("PortProtocols", value, _map_PortProtocols)

_alias_PortProtocols = "set of str"

_map_AllProtocols = {
    "any":  FuelSet(),
    "icmp": FuelSet(["icmp"]),
    "igmp": FuelSet(["igmp"]),
    "tcp":  FuelSet(["tcp"]),
    "egp":  FuelSet(["egp"]),
    "udp":  FuelSet(["udp"]),
    "rsvp": FuelSet(["rsvp"]),
    "esp":  FuelSet(["esp"]),
    "ah":   FuelSet(["ah"]),
    "mtp":  FuelSet(["mtp"]),
    "pim":  FuelSet(["pim"]),
}

def _convert_AllProtocols(value):
    return _check_enum("AnyProtocol", value, _map_AllProtocols)

_alias_AllProtocols = "set of str"

_map_InterfaceType = {
    "line":     PoliWall.MY_IP_LINE,
    "wireless": PoliWall.MY_IP_WIRELESS,
    "all":      PoliWall.MY_IP,
    }

def _convert_InterfaceType(value):
    return _check_enum("InterfaceType", value, _map_InterfaceType)

_alias_InterfaceType = "IPAddr"

def _convert_PortSet(value):
    value = _convert_set_of_int(value)
    if value:
        lo = min(value)
        hi = max(value)
        if lo < 0:
            raise BadValue("Can't have negative port numbers")
        if hi > 65535:
            raise BadValue("Maximum port number is 65535")
    return value

_alias_PortSet = "set of int"

_convert_Target = _convert_set_of_IPAddr
_alias_Target = "set of IPAddr"


_map_ShouldOrNot = {
    "should": True,
    "should not": False,
    }

def _convert_ShouldOrNot(value):
    return _check_enum("ShouldOrNot", value, _map_ShouldOrNot)

_alias_ShouldOrNot = "bool"

def _invert_ShouldOrNot(value):
    if value == "true":
        return "should"
    if value == "false":
        return "should not"
    return value # ???

_map_OSType = {
    "Windows": FuelSet(["windows", "win2k", "win2ks", "win2kws", "winxp", "win2003s"]),
    "Windows 2000": FuelSet(["windows", "win2k", "win2ks", "win2kws"]),
    "Windows XP": FuelSet(["windows", "winxp"]),
    "Windows 2003": FuelSet(["windows", "win2003s"]),
    }

def _convert_OSType(value):
    return _check_enum("OSType", value, _map_OSType)

_alias_OSType = "set of str"

_map_CompOp = {
    "<": FuelSet(["<"]),
    ">": FuelSet([">"]),
    "=": FuelSet(["="]),
    "<=": FuelSet(["<="]),
    ">=": FuelSet([">="]),
    "!=": FuelSet(["!="]),
    }

def _convert_CompOp(value):
    return _check_enum("CompOp", value, _map_CompOp)

_alias_CompOp = "set of str"


def _invert_PortProtocols(value):
    value = value.replace("'", " ").replace('"', " ")
    value = value.replace("{", " ").replace("}", " ")
    value = value.replace(",", " ")
    values = value.split()
    values.sort()
    if values == ["tcp", "udp"]:
        return "both"
    if values == ["udp"]:
        return "udp"
    if values == ["tcp"]:
        return "tcp"
    raise BadValue("Can't invert PortProtocols value %r" % value)

def _invert_Action(value):
    value = value.replace("'", " ").replace('"', " ")
    value = value.replace("{", " ").replace("}", " ")
    value = value.replace(",", " ")
    values = value.split()
    values.sort()
    if values == ["B", "D"]:
        return "deny"
    if values == ["D"]:
        return "deny_silently"
    if values == ["A"]:
        return "allow"
    raise BadValue("Can't invert Action value %r" % value)

def _invert_Target(value):
    if value == "nil":
        return None
    return value

def _invert_InterfaceType(value):
    if value == "nil":
        return None
    return value

_invmap = {}
_convmap = {}
_aliasmap = {}
for _name, _value in list(globals().items()):
    if _name.startswith("_convert_"):
        _convmap[_name[9:].replace("_", " ")] = _value
    elif _name.startswith("_invert_"):
        _invmap[_name[8:].replace("_", " ")] = _value
    elif _name.startswith("_alias_"):
        _aliasmap[_name[7:].replace("_", " ")] = _value

def isvalidtype(typename):
    return typename in _convmap

def convert(value, typename):
    if not isinstance(value, str):
        raise TypeError("convert() value argument should be a string (%.40s)"
                        % repr(value))
    converter = _convmap.get(typename)
    if converter is None:
        raise BadTypeName(typename)
    try:
        return converter(value)
    except ValueError as err:
        raise BadValue(str(err))

# For some types we need to know to which "real" type they alias

def getalias(typename):
    if typename not in _convmap:
        raise BadTypeName(typename)
    return _aliasmap.get(typename, typename)

# For some types we need the inverse, too

def invert(value, typename):
    if not isinstance(value, str):
        raise TypeError("invert() value argument should be a string (%.40s)"
                        % repr(value))
    if typename not in _convmap:
        raise BadTypeName(typename)
    inverter = _invmap.get(typename)
    if inverter is None:
        return value # Default inversion is 1-1
    try:
        return inverter(value)
    except ValueError as err:
        raise BadValue(str(err))
