Normal
  $   ˆl–ßi~”jn-jq@»q·®21hßÂ„l6ßÁÀ¿ ?÷     # -*- coding: utf-8 -*-
#
# SPDX-License-Identifier: GPL-2.0-or-later

__all__ = [ "Policy", "policy_reader", "policy_writer" ]

import xml.sax as sax
import os
import io
import shutil

from firewall import config
from firewall.functions import (
    checkIP,
    checkIP6,
    checkUINT16,
    coalescePortRange,
    max_policy_name_len,
    portInPortRange,
    portStr,
    uniqify,
)
from firewall.core.base import DEFAULT_POLICY_TARGET, POLICY_TARGETS, DEFAULT_POLICY_PRIORITY
from firewall.core.io.io_object import IO_Object, \
    IO_Object_ContentHandler, IO_Object_XMLGenerator, check_port, \
    check_tcpudp, check_protocol
from firewall.core import rich
from firewall.core.logger import log
from firewall import errors
from firewall.errors import FirewallError


def common_startElement(obj, name, attrs):
    if name == "short":
        pass
    elif name == "description":
        pass

    elif name == "service":
        if obj._rule:
            if obj._rule.element:
                log.warning("Invalid rule: More than one element in rule '%s', ignoring.",
                            str(obj._rule))
                obj._rule_error = True
                return True
            obj._rule.element = rich.Rich_Service(attrs["name"])
            return True
        if attrs["name"] not in obj.item.services:
            obj.item.services.append(attrs["name"])
        else:
            log.warning("Service '%s' already set, ignoring.",
                        attrs["name"])

    elif name == "port":
        if obj._rule:
            if obj._rule.element:
                log.warning("Invalid rule: More than one element in rule '%s', ignoring.",
                            str(obj._rule))
                obj._rule_error = True
                return True
            obj._rule.element = rich.Rich_Port(attrs["port"],
                                                attrs["protocol"])
            return True
        check_port(attrs["port"])
        check_tcpudp(attrs["protocol"])

        # coalesce and warn about overlapping ranges
        new_port_id = (portStr(attrs["port"], "-"), attrs["protocol"])
        existing_port_ids = list(filter(lambda x: x[1] == attrs["protocol"], obj.item.ports))
        for port_id in existing_port_ids:
            if portInPortRange(new_port_id[0], port_id[0]):
                # the range is wholly contained already, so just warn
                _name = obj.item.derived_from_zone if isinstance(obj.item, Policy) else obj.item.name
                log.warning(FirewallError(errors.ALREADY_ENABLED,
                                    "'%s:%s' already in '%s'" % (new_port_id[0], attrs["protocol"], _name)))
                break # for
        else:
            # the range can be coalesced into the existing set
            added_ranges, removed_ranges = coalescePortRange(new_port_id[0], [_port for (_port, _protocol) in existing_port_ids])

            for _range in removed_ranges:
                entry = (portStr(_range, "-"), attrs["protocol"])
                obj.item.ports.remove(entry)
            for _range in added_ranges:
                entry = (portStr(_range, "-"), attrs["protocol"])
                obj.item.ports.append(entry)

    elif name == "protocol":
        if obj._rule:
            if obj._rule.element:
                log.warning("Invalid rule: More than one element in rule '%s', ignoring.",
                            str(obj._rule))
                obj._rule_error = True
                return True
            obj._rule.element = rich.Rich_Protocol(attrs["value"])
        else:
            check_protocol(attrs["value"])
            if attrs["value"] not in obj.item.protocols:
                obj.item.protocols.append(attrs["value"])
            else:
                log.warning("Protocol '%s' already set, ignoring.",
                            attrs["value"])

    elif name == "tcp-mss-clamp":
        if obj._rule:
            if obj._rule.element:
                log.warning("Invalid rule: More than one element in rule