403WebShell

403Webshell
Server IP : 13.127.148.211 / Your IP : 216.73.216.13
Web Server : Apache/2.4.41 (Ubuntu)
System : Linux ip-172-31-43-195 5.15.0-1084-aws #91~20.04.1-Ubuntu SMP Fri May 2 06:59:36 UTC 2025 x86_64
User : www-data ( 33)
PHP Version : 7.4.3-4ubuntu2.29
Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare,
MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON
Directory : /usr/lib/python3/dist-packages/ufw/
Upload File :
[ Back ]
Current File : /usr/lib/python3/dist-packages/ufw/parser.py
#
# parser.py: parser class for ufw
#
# Copyright 2009-2018 Canonical Ltd.
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU General Public License version 3,
#    as published by the Free Software Foundation.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
#    You should have received a copy of the GNU General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
#
# Adding New Commands
#
# 1. Create a new UFWCommandFoo object that implements UFWCommand
# 2. Create UFWCommandFoo.parse() to return a UFWParserResponse object
# 3. Create UFWCommandFoo.help() to display help for this command
# 4. Register this command with the parser using:
#    parser.register_command(UFWCommandFoo('foo'))
#
#
# Extending Existing Commands
#
# 1. Register the new command with an existing UFWCommand via
#    register_command(). Eg
#    parser.register_command(UFWCommandNewcommand('new_command'))
# 2. Update UFWCommandExisting.parse() for new_command
# 3. Update UFWCommandExisting.help() for new_command
#

import re
import ufw.util
import ufw.applications
from ufw.common import UFWError
from ufw.util import debug


class UFWCommand:
    '''Generic class for parser commands.'''
    def __init__(self, type, command):
        self.command = command
        self.types = []
        if type not in self.types:
            self.types.append(type)
        self.type = type

    def parse(self, argv):
        if len(argv) < 1:
            raise ValueError()

        r = UFWParserResponse(argv[0].lower())

        return r

    def help(self, args):
        raise UFWError("UFWCommand.help: need to override")


class UFWCommandRule(UFWCommand):
    '''Class for parsing ufw rule commands'''
    def __init__(self, command):
        type = 'rule'
        UFWCommand.__init__(self, type, command)

    def parse(self, argv):
        action = ""
        rule = ""
        type = ""
        from_type = "any"
        to_type = "any"
        from_service = ""
        to_service = ""
        insert_pos = ""
        logtype = ""
        remove = False

        if len(argv) > 0 and argv[0].lower() == "rule":
            argv.remove(argv[0])

        # TODO: break this out
        if len(argv) > 0:
            if argv[0].lower() == "delete" and len(argv) > 1:
                remove = True
                argv.remove(argv[0])
                rule_num = None
                try:
                    rule_num = int(argv[0])
                except Exception:
                    action = argv[0]

                # return quickly if deleting by rule number
                if rule_num is not None:
                    r = UFWParserResponse('delete-%d' % rule_num)
                    return r

            elif argv[0].lower() == "insert":
                if len(argv) < 4:
                    raise ValueError()
                insert_pos = argv[1]

                # Using position '0' appends the rule while '-1' prepends,
                # which is potentially confusing for the end user
                if insert_pos == "0" or insert_pos == "-1":
                    err_msg = _("Cannot insert rule at position '%s'") % \
                                (insert_pos)
                    raise UFWError(err_msg)

                # strip out 'insert NUM' and parse as normal
                del argv[1]
                del argv[0]

            elif argv[0].lower() == "prepend":
                insert_pos = -1
                del argv[0]

            action = argv[0]

        if action != "allow" and action != "deny" and action != "reject" and \
           action != "limit":
            raise ValueError()

        nargs = len(argv)
        if nargs < 2:
            raise ValueError()

        # set/strip
        rule_direction = "in"
        if nargs > 1 and (argv[1].lower() == "in" or \
                          argv[1].lower() == "out"):
            rule_direction = argv[1].lower()

        # strip out direction if not an interface rule
        if nargs > 2 and argv[2] != "on" and (argv[1].lower() == "in" or \
                                              argv[1].lower() == "out"):
            rule_direction = argv[1].lower()
            del argv[1]
            nargs = len(argv)

        # strip out 'on' as in 'allow in on eth0 ...'
        has_interface = False
        if nargs > 1 and (argv.count('in') > 0 or argv.count('out') > 0):
            err_msg = _("Invalid interface clause")

            if argv[1].lower() != "in" and argv[1].lower() != "out":
                raise UFWError(err_msg)
            if nargs < 3 or argv[2].lower() != "on":
                raise UFWError(err_msg)

            del argv[2]
            nargs = len(argv)
            has_interface = True

        log_idx = 0
        if has_interface and nargs > 3 and (argv[3].lower() == "log" or \
                                            argv[3].lower() == 'log-all'):
            log_idx = 3
        elif nargs > 2 and (argv[1].lower() == "log" or \
                           argv[1].lower() == 'log-all'):
            log_idx = 1

        if log_idx > 0:
            logtype = argv[log_idx].lower()
            # strip out 'log' or 'log-all' and parse as normal
            del argv[log_idx]
            nargs = len(argv)

        if "log" in argv:
            err_msg = _("Option 'log' not allowed here")
            raise UFWError(err_msg)

        if "log-all" in argv:
            err_msg = _("Option 'log-all' not allowed here")
            raise UFWError(err_msg)

        comment = ""
        if 'comment' in argv:
            comment_idx = argv.index("comment")
            if comment_idx == len(argv) - 1:
                err_msg = _("Option 'comment' missing required argument")
                raise UFWError(err_msg)
            comment = argv[comment_idx+1]
            # TODO: properly support "'" in the comment string. See r949 for
            # details
            if "'" in comment:
                err_msg = _("Comment may not contain \"'\"")
                raise ValueError(err_msg)

            del argv[comment_idx+1]
            del argv[comment_idx]
            nargs = len(argv)

        if nargs < 2 or nargs > 13:
            raise ValueError()

        rule_action = action
        if logtype != "":
            rule_action += "_" + logtype
        rule = ufw.common.UFWRule(rule_action, "any", "any", \
                                  direction=rule_direction,
                                  comment=ufw.util.hex_encode(comment))
        if remove:
            rule.remove = remove
        elif insert_pos != "":
            try:
                rule.set_position(insert_pos)
            except Exception:
                raise
        if nargs == 2:
            # Short form where only app or port/proto is given
            if ufw.applications.valid_profile_name(argv[1]):
                # Check if name collision with /etc/services. If so, use
                # /etc/services instead of application profile
                try:
                    ufw.util.get_services_proto(argv[1])
                except Exception:
                    type = "both"
                    rule.dapp = argv[1]
                    rule.set_port(argv[1], "dst")
            if rule.dapp == "":
                try:
                    (port, proto) = ufw.util.parse_port_proto(argv[1])
                except ValueError as e:
                    raise UFWError(e)

                if not re.match('^\d([0-9,:]*\d+)*$', port):
                    if ',' in port or ':' in port:
                        err_msg = _("Port ranges must be numeric")
                        raise UFWError(err_msg)
                    to_service = port

                try:
                    rule.set_protocol(proto)
                    rule.set_port(port, "dst")
                    type = "both"
                except UFWError:
                    err_msg = _("Bad port")
                    raise UFWError(err_msg)
        elif (nargs + 1) % 2 != 0:
            err_msg = _("Wrong number of arguments")
            raise UFWError(err_msg)
        elif 'from' not in argv and 'to' not in argv and 'in' not in argv and \
             'out' not in argv:
            err_msg = _("Need 'to' or 'from' clause")
            raise UFWError(err_msg)
        else:
            # Full form with PF-style syntax
            keys = [ 'proto', 'from', 'to', 'port', 'app', 'in', 'out' ]

            # quick check
            if argv.count("to") > 1 or \
               argv.count("from") > 1 or \
               argv.count("proto") > 1 or \
               argv.count("port") > 2 or \
               argv.count("in") > 1 or \
               argv.count("out") > 1 or \
               argv.count("app") > 2 or \
               argv.count("app") > 0 and argv.count("proto") > 0:
                err_msg = _("Improper rule syntax")
                raise UFWError(err_msg)

            i = 0
            loc = ""
            for arg in argv:
                if i % 2 != 0 and argv[i] not in keys:
                    err_msg = _("Invalid token '%s'") % (argv[i])
                    raise UFWError(err_msg)
                if arg == "proto":
                    if i+1 < nargs:
                        try:
                            rule.set_protocol(argv[i+1])
                        except Exception:
                            raise
                    else: # pragma: no cover
                        # This can't normally be reached because of nargs
                        # checks above, but leave it here in case our parsing
                        # changes
                        err_msg = _("Invalid 'proto' clause")
                        raise UFWError(err_msg)
                elif arg == "in" or arg == "out":
                    if i+1 < nargs:
                        try:
                            if arg == "in":
                                rule.set_interface("in", argv[i+1])
                            elif arg == "out":
                                rule.set_interface("out", argv[i+1])
                        except Exception:
                            raise
                    else: # pragma: no cover
                        # This can't normally be reached because of nargs
                        # checks above, but leave it here in case our parsing
                        # changes
                        err_msg = _("Invalid '%s' clause") % (arg)
                        raise UFWError(err_msg)
                elif arg == "from":
                    if i+1 < nargs:
                        try:
                            faddr = argv[i+1].lower()
                            if faddr == "any":
                                faddr = "0.0.0.0/0"
                                from_type = "any"
                            else:
                                if ufw.util.valid_address(faddr, "6"):
                                    from_type = "v6"
                                else:
                                    from_type = "v4"
                            rule.set_src(faddr)
                        except Exception:
                            raise
                        loc = "src"
                    else: # pragma: no cover
                        # This can't normally be reached because of nargs
                        # checks above, but leave it here in case our parsing
                        # changes
                        err_msg = _("Invalid 'from' clause")
                        raise UFWError(err_msg)
                elif arg == "to":
                    if i+1 < nargs:
                        try:
                            saddr = argv[i+1].lower()
                            if saddr == "any":
                                saddr = "0.0.0.0/0"
                                to_type = "any"
                            else:
                                if ufw.util.valid_address(saddr, "6"):
                                    to_type = "v6"
                                else:
                                    to_type = "v4"
                            rule.set_dst(saddr)
                        except Exception:
                            raise
                        loc = "dst"
                    else: # pragma: no cover
                        # This can't normally be reached because of nargs
                        # checks above, but leave it here in case our parsing
                        # changes
                        err_msg = _("Invalid 'to' clause")
                        raise UFWError(err_msg)
                elif arg == "port" or arg == "app":
                    if i+1 < nargs:
                        if loc == "":
                            err_msg = _("Need 'from' or 'to' with '%s'") % \
                                        (arg)
                            raise UFWError(err_msg)

                        tmp = argv[i+1]
                        if arg == "app":
                            if loc == "src":
                                rule.sapp = tmp
                            else:
                                rule.dapp = tmp
                        elif not re.match('^\d([0-9,:]*\d+)*$', tmp):
                            if ',' in tmp or ':' in tmp:
                                err_msg = _("Port ranges must be numeric")
                                raise UFWError(err_msg)

                            if loc == "src":
                                from_service = tmp
                            else:
                                to_service = tmp
                        try:
                            rule.set_port(tmp, loc)
                        except Exception:
                            raise
                    else: # pragma: no cover
                        # This can't normally be reached because of nargs
                        # checks above, but leave it here in case our parsing
                        # changes
                        err_msg = _("Invalid 'port' clause")
                        raise UFWError(err_msg)
                i += 1

            # Figure out the type of rule (IPv4, IPv6, or both) this is
            if from_type == "any" and to_type == "any":
                type = "both"
            elif from_type != "any" and to_type != "any" and \
                 from_type != to_type:
                err_msg = _("Mixed IP versions for 'from' and 'to'")
                raise UFWError(err_msg)
            elif from_type != "any":
                type = from_type
            elif to_type != "any":
                type = to_type

        # Adjust protocol
        if to_service != "" or from_service != "":
            proto = ""
            if to_service != "":
                try:
                    proto = ufw.util.get_services_proto(to_service)
                except Exception: # pragma: no cover
                    # This can't normally be reached because of set_port()
                    # checks above, but leave it here in case our parsing
                    # changes
                    err_msg = _("Could not find protocol")
                    raise UFWError(err_msg)
            if from_service != "":
                if proto == "any" or proto == "":
                    try:
                        proto = ufw.util.get_services_proto(from_service)
                    except Exception: # pragma: no cover
                        # This can't normally be reached because of set_port()
                        # checks above, but leave it here in case our parsing
                        # changes
                        err_msg = _("Could not find protocol")
                        raise UFWError(err_msg)
                else:
                    try:
                        tmp = ufw.util.get_services_proto(from_service)
                    except Exception: # pragma: no cover
                        # This can't normally be reached because of set_port()
                        # checks above, but leave it here in case our parsing
                        # changes
                        err_msg = _("Could not find protocol")
                        raise UFWError(err_msg)
                    if proto == "any" or proto == tmp:
                        proto = tmp
                    elif tmp == "any":
                        pass
                    else:
                        err_msg = _("Protocol mismatch (from/to)")
                        raise UFWError(err_msg)

            # Verify found proto with specified proto
            if rule.protocol == "any":
                rule.set_protocol(proto)
            elif proto != "any" and rule.protocol != proto:
                err_msg = _("Protocol mismatch with specified protocol %s") % \
                            (rule.protocol)
                raise UFWError(err_msg)

        # adjust type as needed
        if rule:
            if rule.protocol in ufw.util.ipv4_only_protocols and \
               type == "both":
                debug("Adjusting iptype to 'v4' for protocol '%s'" % \
                      (rule.protocol))
                type = "v4"

            # Now verify the rule
            rule.verify(type)

        r = UFWParserResponse(action)
        r.data['type'] = self.type
        r.data['rule'] = rule
        r.data['iptype'] = type

        return r

    def get_command(r):
        '''Get command string for rule'''
        res = r.action

        if (r.dst == "0.0.0.0/0" or r.dst == "::/0") and \
           (r.src == "0.0.0.0/0" or r.src == "::/0") and \
           r.sport == "any" and \
           r.sapp == "" and \
           r.interface_in == "" and \
           r.interface_out == "" and \
           r.dport != "any":
            # Short syntax
            if r.direction == "out":
                res += " %s" % r.direction
            if r.logtype != "":
                res += " %s" % r.logtype
            if r.dapp != "":
                if " " in r.dapp:
                    res += " '%s'" % r.dapp
                else:
                    res += " %s" % r.dapp
            else:
                res += " %s" % r.dport
                if r.protocol != "any":
                    res += "/%s" % r.protocol
            if r.comment != "":
                res += " comment '%s'" % r.get_comment()
        else:
            # Full syntax
            if r.interface_in != "":
                res += " in on %s" % r.interface_in
            if r.interface_out != "":
                res += " out on %s" % r.interface_out
            elif r.direction == "out":
                res += " %s" % r.direction
            if r.logtype != "":
                res += " %s" % r.logtype

            for i in ['src', 'dst']:
                if i == 'src':
                    loc = r.src
                    port = r.sport
                    app = r.sapp
                    dir = "from"
                else:
                    loc = r.dst
                    port = r.dport
                    app = r.dapp
                    dir = "to"

                if loc == "0.0.0.0/0" or loc == "::/0":
                    loc = "any"

                if loc != "any" or port != "any" or app != "":
                    res += " %s %s" % (dir, loc)
                    if app != "":
                        if " " in app:
                            res += " app '%s'" % app
                        else:
                            res += " app %s" % app
                    elif port != "any":
                        res += " port %s" % port

            # If still haven't added more than action, direction and/or
            # logtype, then we have a very generic rule, so add 'to any' to
            # mark it as extended form.
            if ' to ' not in res and ' from ' not in res and \
                    r.interface_in == "" and r.interface_out == "":
                res += " to any"

            if r.protocol != "any" and r.dapp == "" and r.sapp == "":
                res += " proto %s" % r.protocol

            if r.comment != "":
                res += " comment '%s'" % r.get_comment()

        return res
    get_command = staticmethod(get_command)


class UFWCommandRouteRule(UFWCommandRule):
    '''Class for parsing ufw route rule commands'''
    def __init__(self, command):
        UFWCommandRule.__init__(self, command)
        self.type = 'route'

    def parse(self, argv):
        assert(argv[0] == "route")

        # 'ufw delete NUM' is the correct usage, not 'ufw route delete NUM'
        if 'delete' in argv:
            idx = argv.index('delete')
            err_msg = ""
            if len(argv) > idx:
                try:
                    # 'route delete NUM' is unsupported
                    int(argv[idx + 1])
                    err_msg = _("'route delete NUM' unsupported. Use 'delete NUM' instead.")
                    raise UFWError(err_msg)
                except ValueError:
                    # 'route delete RULE' is supported
                    pass

        # Let's use as much as UFWCommandRule.parse() as possible. The only
        # difference with our rules is that argv[0] is 'route' and we support
        # both 'in on <interface>' and 'out on <interface>' in our rules.
        # Because UFWCommandRule.parse() expects that the interface clause is
        # specified first, strip out the second clause and add it later
        rule_argv = None
        interface = None
        strip = None

        # eg: ['route', 'allow', 'in', 'on', 'eth0', 'out', 'on', 'eth1']
        s = " ".join(argv)
        if " in on " in s and " out on " in s:
            strip = "out"
            if argv.index("in") > argv.index("out"):
                strip = "in"
            # Remove 2nd interface clause from argv and add it to the rule
            # later. Because we searched for " <strip> on " in our joined
            # string we are guaranteed to have argv[argv.index(<strip>) + 2]
            # exist.
            interface = argv[argv.index(strip) + 2]
            rule_argv = argv[0:argv.index(strip)] + argv[argv.index(strip)+3:]
        elif not re.search(r' (in|out) on ', s) and \
             not re.search(r' app (in|out) ', s) and \
             (" in " in s or " out " in s):
            # Specifying a direction without an interface doesn't make any
            # sense with route rules. application names could be 'in' or 'out'
            # so don't artificially limit those names.
            err_msg = _("Invalid interface clause for route rule")
            raise UFWError(err_msg)
        else:
            rule_argv = argv

        rule_argv[0] = "rule"
        r = UFWCommandRule.parse(self, rule_argv)
        if 'rule' in r.data:
            r.data['rule'].forward = True
            if strip and interface:
                r.data['rule'].set_interface(strip, interface)

        return r


class UFWCommandApp(UFWCommand):
    '''Class for parsing ufw application commands'''
    def __init__(self, command):
        type = 'app'
        UFWCommand.__init__(self, type, command)

    def parse(self, argv):
        '''Parse applications command.'''
        name = ""
        action = ""
        addnew = False

        if argv[0] != "app":
            raise ValueError()
        del argv[0]

        nargs = len(argv)
        action = argv[0].lower()

        if action == "info" or action == "update":
            if nargs >= 3 and argv[1] == "--add-new":
                addnew = True
                argv.remove("--add-new")
                nargs = len(argv)

            if nargs < 2:
                raise ValueError()

            # Handle quoted name with spaces in it by stripping Python's ['...']
            # list as string text.
            name = str(argv[1]).strip("[']")

            if addnew:
                action += "-with-new"

        if action == "list" and nargs != 1:
            raise ValueError()

        if action == "default":
            if nargs < 2:
                raise ValueError()
            if argv[1].lower() == "allow":
                action = "default-allow"
            elif argv[1].lower() == "deny":
                action = "default-deny"
            elif argv[1].lower() == "reject":
                action = "default-reject"
            elif argv[1].lower() == "skip":
                action = "default-skip"
            else:
                raise ValueError()

        r = UFWParserResponse(action)
        r.data['type'] = self.type
        r.data['name'] = name

        return r


class UFWCommandBasic(UFWCommand):
    '''Class for parsing ufw basic commands'''
    def __init__(self, command):
        type = 'basic'
        UFWCommand.__init__(self, type, command)

    def parse(self, argv):
        if len(argv) != 1:
            raise ValueError()
        return UFWCommand.parse(self, argv)


class UFWCommandDefault(UFWCommand):
    '''Class for parsing ufw default commands'''
    def __init__(self, command):
        type = 'default'
        UFWCommand.__init__(self, type, command)

    def parse(self, argv):
        # Basic sanity check
        if len(argv) < 2:
            raise ValueError()

        # Set the direction
        action = ""
        direction = "incoming"
        if len(argv) > 2:
            if argv[2].lower() != "incoming" and \
               argv[2].lower() != "input" and \
               argv[2].lower() != "routed" and \
               argv[2].lower() != "forward" and \
               argv[2].lower() != "output" and \
               argv[2].lower() != "outgoing":
                raise ValueError()
            if argv[2].lower().startswith("in"):
                direction = "incoming"
            elif argv[2].lower().startswith("out"):
                direction = "outgoing"
            elif argv[2].lower() == "routed" or argv[2].lower() == "forward":
                direction = "routed"
            else:  # pragma: no cover
                direction = argv[2].lower()

        # Set the policy
        if argv[1].lower() == "deny":
            action = "default-deny"
        elif argv[1].lower() == "allow":
            action = "default-allow"
        elif argv[1].lower() == "reject":
            action = "default-reject"
        else:
            raise ValueError()

        action += "-%s" % (direction)

        return UFWParserResponse(action)


class UFWCommandLogging(UFWCommand):
    '''Class for parsing ufw logging commands'''
    def __init__(self, command):
        type = 'logging'
        UFWCommand.__init__(self, type, command)

    def parse(self, argv):
        action = ""
        if len(argv) < 2:
            raise ValueError()
        elif argv[1].lower() == "off":
            action = "logging-off"
        elif argv[1].lower() == "on" or argv[1].lower() == "low" or \
             argv[1].lower() == "medium" or argv[1].lower() == "high" or \
             argv[1].lower() == "full":
            action = "logging-on"
            if argv[1].lower() != "on":
                action += "_" + argv[1].lower()
        else:
            raise ValueError()

        return UFWParserResponse(action)


class UFWCommandStatus(UFWCommand):
    '''Class for parsing ufw status commands'''
    def __init__(self, command):
        type = 'status'
        UFWCommand.__init__(self, type, command)

    def parse(self, argv):
        r = UFWCommand.parse(self, argv)
        if len(argv) == 1:
            r.action = "status"
        elif len(argv) > 1:
            if argv[1].lower() == "verbose":
                r.action = "status-verbose"
            elif argv[1].lower() == "numbered":
                r.action = "status-numbered"
            else:
                raise ValueError()
        return r


class UFWCommandShow(UFWCommand):
    '''Class for parsing ufw show commands'''
    def __init__(self, command):
        type = 'show'
        UFWCommand.__init__(self, type, command)

    def parse(self, argv):
        action = ""
        if len(argv) == 1:
            raise ValueError()
        elif argv[1].lower() == "raw":
            action = "show-raw"
        elif argv[1].lower() == "before-rules":
            action = "show-before"
        elif argv[1].lower() == "user-rules":
            action = "show-user"
        elif argv[1].lower() == "after-rules":
            action = "show-after"
        elif argv[1].lower() == "logging-rules":
            action = "show-logging"
        elif argv[1].lower() == "builtins":
            action = "show-builtins"
        elif argv[1].lower() == "listening":
            action = "show-listening"
        elif argv[1].lower() == "added":
            action = "show-added"
        else:
            raise ValueError()

        return UFWParserResponse(action)


class UFWParserResponse:
    '''Class for ufw parser response'''
    def __init__(self, action):
        self.action = action.lower()
        self.dryrun = False
        self.force = False
        self.data = {}

    def __str__(self):
        s = "action='%s'" % (self.action)
        keys = list(self.data.keys())
        keys.sort()
        for i in keys:
            s += ",%s='%s'" % (i, self.data[i])
        s += "\n"

        return repr(s)


class UFWParser:
    '''Class for ufw parser'''
    def __init__(self):
        self.commands = {}

    def allowed_command(self, type, cmd):
        '''Return command if it is allowed, otherwise raise an exception'''
        if type.lower() not in list(self.commands.keys()):
            raise ValueError()

        if cmd.lower() not in list(self.commands[type].keys()):
            raise ValueError()

        return cmd.lower()

    def parse_command(self, args):
        '''Parse command. Returns a UFWParserAction'''
        dryrun = False
        if len(args) > 0 and args[0].lower() == "--dry-run":
            dryrun = True
            args.remove(args[0])

        force = False
        if len(args) > 0 and (args[0].lower() == "--force" or \
                              args[0].lower() == "-f"):
            force = True
            args.remove(args[0])

        cmd = ""
        type = ""

        tmp = args[0].lower()
        if len(args) > 1 and tmp in list(self.commands.keys()) and \
                args[1].lower() in list(self.commands[tmp].keys()):
            type = tmp
            cmd = args[1].lower()
        else:
            # Discover the type
            cmd = tmp
            for i in list(self.commands.keys()):
                if cmd in self.commands[i]:
                    # Skip any inherited commands that inherit from
                    # UFWCommandRule since they must have more than one
                    # argument to be valid and used
                    if isinstance(self.commands[i][cmd], UFWCommandRule) and \
                       getattr(self.commands[i][cmd], 'type') != 'rule':
                        continue  # pragma: nocover
                    type = i
                    break
            if type == "":
                type = 'rule'

        action = self.allowed_command(type, cmd)

        cmd = self.commands[type][action]
        response = cmd.parse(args)
        response.dryrun = dryrun
        response.force = force

        return response

    def register_command(self, c):
        '''Register a command with the parser'''
        if c.command is None or c.command == '':
            # If the command is empty, then use 'type' as command
            key = "%s" % (c.type)
        else:
            key = "%s" % (c.command)

        if c.type not in self.commands:
            self.commands[c.type] = {}
        if key in self.commands[c.type]:
            err_msg = _("Command '%s' already exists") % (key)
            raise UFWError(err_msg)
        self.commands[c.type][key] = c
Youez - 2016 - github.com/yon3zu
LinuXploit