File: //bin/firewall-offline-cmd
#!/usr/bin/python3
# SPDX-License-Identifier: GPL-2.0-or-later
#
# Copyright (C) 2009-2016 Red Hat, Inc.
#
# Authors:
# Thomas Woerner <twoerner@redhat.com>
# Jiri Popelka <jpopelka@redhat.com>
from gi.repository import GObject
import sys
sys.modules["gobject"] = GObject
import argparse
import os
from firewall.client import (
FirewallClientIPSetSettings,
FirewallClientZoneSettings,
FirewallClientServiceSettings,
FirewallClientIcmpTypeSettings,
FirewallClientHelperSettings,
FirewallClientPolicySettings,
)
from firewall.errors import FirewallError
from firewall import config
from firewall.core.fw import Firewall
from firewall.functions import joinArgs, splitArgs, getPortRange
from firewall.core.io.functions import check_on_disk_config
from firewall.core.io.zone import zone_reader
from firewall.core.io.policy import policy_reader
from firewall.core.io.service import service_reader
from firewall.core.io.ipset import ipset_reader
from firewall.core.io.icmptype import icmptype_reader
from firewall.core.io.helper import helper_reader
from firewall.command import FirewallCommand
# check for root user
def assert_root():
if os.getuid() != 0:
sys.stderr.write("You need to be root to run %s.\n" % sys.argv[0])
sys.exit(-1)
SYSTEM_CONFIG_FIREWALL = config.SYSCONFIGDIR + "/system-config-firewall"
def __usage():
sys.stdout.write(
"""
Usage: firewall-offline-cmd [OPTIONS...]
If no options are given, configuration from '%s' will be migrated.
General Options
-h, --help Prints a short help text and exits
-V, --version Print the version string of firewalld
-q, --quiet Do not print status messages
--system-config Path to firewalld system configuration
--default-config Path to firewalld default configuration
--check-config Check system and default configuration
--reset-to-defaults
Reset configuration to firewalld's default configuration
Lokkit Compatibility Options
--migrate-system-config-firewall=<file>
Import configuration data from the given configuration
file.
--enabled Enable firewall (default)
--disabled Disable firewall
--addmodule=<module> Ignored option, was used to enable an iptables module
--removemodule=<module>
Ignored option, was used to disable an iptables module
-s <service>, --service=<service>
Enable a service in the default zone (example: ssh)
--remove-service=<service>
Disable a service in the default zone (example: ssh)
-p <port>[-<port>]:<protocol>, --port=<port>[-<port>]:<protocol>
Enable a port in the default zone (example: ssh:tcp)
-t <interface>, --trust=<interface>
Bind an interface to the trusted zone
-m <interface>, --masq=<interface>
Enables masquerading in the default zone, interface
argument is ignored. This is IPv4 only.
--custom-rules=[<type>:][<table>:]<filename>
Ignored option. Was used to add custom rules to the
firewall (Example:
ipv4:filter:%s/ipv4_filter_addon)
--forward-port=if=<interface>:port=<port>:proto=<protocol>[:toport=<destination port>][:toaddr=<destination address>]
Forward the port with protocol for the interface to
either another local destination port (no destination
address given) or to an other destination address with
an optional destination port. This will be added to
the default zone. This is IPv4 only.
--block-icmp=<icmp type>
Block this ICMP type in the default zone. The default
is to accept all ICMP types.
Log Denied Options
--get-log-denied Print the log denied value
--set-log-denied=<value>
Set log denied value
Automatic Helpers Options
--get-automatic-helpers
Print the automatic helpers value
--set-automatic-helpers=<value>
Set automatic helpers value
Zone Options
--get-default-zone Print default zone for connections and interfaces
--set-default-zone=<zone>
Set default zone
--get-zones Print predefined zones
--get-services Print predefined services
--get-icmptypes Print predefined icmptypes
--get-zone-of-interface=<interface>
Print name of the zone the interface is bound to
--get-zone-of-source=<source>[/<mask>]|<MAC>|ipset:<ipset>
Print name of the zone the source is bound to
--list-all-zones List everything added for or enabled in all zones
--new-zone=<zone> Add a new empty zone
--new-zone-from-file=<filename> [--name=<zone>]
Add a new zone from file with optional name override [P only]
--delete-zone=<zone> Delete an existing zone
--load-zone-defaults=<zone>
Load zone default settings
--zone=<zone> Use this zone to set or query options, else default zone
Usable for options marked with [Z]
--info-zone=<zone> Print information about a zone
--path-zone=<zone> Print file path of a zone
Policy Options
--get-policies Print predefined policies
--list-all-policies List everything added for or enabled in all policies
--new-policy=<policy>
Add a new empty policy
--new-policy-from-file=<filename> [--name=<policy>]
Add a new policy from file with optional name override [P only]
--delete-policy=<policy>
Delete an existing policy
--load-policy-defaults=<policy>
Load policy default settings
--policy=<policy> Use this policy to set or query options
Usable for options marked with [O]
--info-policy=<policy>
Print information about a policy
--path-policy=<policy>
Print file path of a policy
IPSet Options
--new-ipset=<ipset> --type=<ipset type> [--option=<key>[=<value>]]..
Add a new empty ipset
--new-ipset-from-file=<filename> [--name=<ipset>]
Add a new ipset from file with optional name override [P only]
--delete-ipset=<ipset>
Delete an existing ipset
--load-ipset-defaults=<ipset>
Load ipset default settings
--info-ipset=<ipset> Print information about an ipset
--path-ipset=<ipset> Print file path of an ipset
--get-ipsets Print predefined ipsets
--ipset=<ipset> --set-description=<description>
Set new description to ipset
--ipset=<ipset> --get-description
Print description for ipset
--ipset=<ipset> --set-short=<description>
Set new short description to ipset
--ipset=<ipset> --get-short
Print short description for ipset
--ipset=<ipset> --add-entry=<entry>
Add a new entry to an ipset
--ipset=<ipset> --remove-entry=<entry>
Remove an entry from an ipset
--ipset=<ipset> --query-entry=<entry>
Return whether ipset has an entry
--ipset=<ipset> --get-entries
List entries of an ipset
--ipset=<ipset> --add-entries-from-file=<entry>
Add a new entries to an ipset
--ipset=<ipset> --remove-entries-from-file=<entry>
Remove entries from an ipset
IcmpType Options
--new-icmptype=<icmptype>
Add a new empty icmptype
--new-icmptype-from-file=<filename> [--name=<icmptype>]
Add a new icmptype from file with optional name override [P only]
--delete-icmptype=<icmptype>
Delete an existing icmptype
--load-icmptype-defaults=<icmptype>
Load icmptype default settings
--info-icmptype=<icmptype>
Print information about an icmptype
--path-icmptype=<icmptype>
Print file path of an icmptype
--icmptype=<icmptype> --set-description=<description>
Set new description to icmptype
--icmptype=<icmptype> --get-description
Print description for icmptype
--icmptype=<icmptype> --set-short=<description>
Set new short description to icmptype
--icmptype=<icmptype> --get-short
Print short description for icmptype
--icmptype=<icmptype> --add-destination=<ipv>
Enable destination for ipv in icmptype
--icmptype=<icmptype> --remove-destination=<ipv>
Disable destination for ipv in icmptype
--icmptype=<icmptype> --query-destination=<ipv>
Return whether destination ipv is enabled in icmptype
--icmptype=<icmptype> --get-destinations
List destinations in icmptype
Service Options
--new-service=<service>
Add a new empty service
--new-service-from-file=<filename> [--name=<service>]
Add a new service from file with optional name override [P only]
--delete-service=<service>
Delete an existing service
--load-service-defaults=<service>
Load icmptype default settings
--info-service=<service>
Print information about a service
--path-service=<service>
Print file path of a service
--service=<service> --set-description=<description>
Set new description to service
--service=<service> --get-description
Print description for service
--service=<service> --set-short=<description>
Set new short description to service
--service=<service> --get-short
Print short description for service
--service=<service> --add-port=<portid>[-<portid>]/<protocol>
Add a new port to service
--service=<service> --remove-port=<portid>[-<portid>]/<protocol>
Remove a port from service
--service=<service> --query-port=<portid>[-<portid>]/<protocol>
Return whether the port has been added for service
--service=<service> --get-ports
List ports of service
--service=<service> --add-protocol=<protocol>
Add a new protocol to service
--service=<service> --remove-protocol=<protocol>
Remove a protocol from service
--service=<service> --query-protocol=<protocol>
Return whether the protocol has been added for service
--service=<service> --get-protocols
List protocols of service
--service=<service> --add-source-port=<portid>[-<portid>]/<protocol>
Add a new source port to service
--service=<service> --remove-source-port=<portid>[-<portid>]/<protocol>
Remove a source port from service
--service=<service> --query-source-port=<portid>[-<portid>]/<protocol>
Return whether the source port has been added for service [P only]
--service=<service> --get-source-ports
List source ports of service
--service=<service> --add-helper=<helper>
Add a new helper to service
--service=<service> --remove-helper=<helper>
Remove a helper from service
--service=<service> --query-helper=<helper>
Return whether the helper has been added for service
--service=<service> --get-service-helpers
List helpers of service
--service=<service> --set-destination=<ipv>:<address>[/<mask>]
Set destination for ipv to address in service
--service=<service> --remove-destination=<ipv>
Disable destination for ipv i service
--service=<service> --query-destination=<ipv>:<address>[/<mask>]
Return whether destination ipv is set for service
--service=<service> --get-destinations
List destinations in service
--service=<service> --add-include=<service>
Add a new include to service
--service=<service> --remove-include=<service>
Remove a include from service
--service=<service> --query-include=<service>
Return whether the include has been added for service
--service=<service> --get-includes
List includes of service
Options to Adapt and Query Zones and Policies
--list-all List everything added for or enabled [Z] [O]
--set-description=<description>
Set new description [Z] [O]
--get-description Print description [Z] [O]
--get-target Get the target [Z] [O]
--set-target=<target>
Set the target [Z] [O]
--set-short=<description>
Set new short description to zone [Z] [O]
--get-short Print short description for zone [Z] [O]
--list-services List services added [Z] [O]
--add-service=<service>
Add a service [Z] [O]
--remove-service-from-zone=<service>
Remove a service from a zone [Z]
--remove-service-from-policy=<service>
Remove a service from a policy [O]
--query-service=<service>
Return whether service has been added [Z] [O]
--list-ports List ports added [Z] [O]
--add-port=<portid>[-<portid>]/<protocol>
Add the port [Z] [O]
--remove-port=<portid>[-<portid>]/<protocol>
Remove the port [Z] [O]
--query-port=<portid>[-<portid>]/<protocol>
Return whether the port has been added [Z] [O]
--list-protocols List protocols added [Z] [O]
--add-protocol=<protocol>
Add the protocol [Z] [O]
--remove-protocol=<protocol>
Remove the protocol [Z] [O]
--query-protocol=<protocol>
Return whether the protocol has been added [Z] [O]
--list-source-ports List source ports added [Z] [O]
--add-source-port=<portid>[-<portid>]/<protocol>
Add the source port [Z] [O]
--remove-source-port=<portid>[-<portid>]/<protocol>
Remove the source port [Z] [O]
--query-source-port=<portid>[-<portid>]/<protocol>
Return whether the source port has been added [Z] [O]
--list-icmp-blocks List Internet ICMP type blocks added [Z] [O]
--add-icmp-block=<icmptype>
Add an ICMP block [Z] [O]
--remove-icmp-block=<icmptype>
Remove the ICMP block [Z] [O]
--query-icmp-block=<icmptype>
Return whether an ICMP block has been added [Z] [O]
--list-forward-ports List IPv4 forward ports added [Z] [O]
--add-forward-port=port=<portid>[-<portid>]:proto=<protocol>[:toport=<portid>[-<portid>]][:toaddr=<address>[/<mask>]]
Add the IPv4 forward port [Z] [O]
--remove-forward-port=port=<portid>[-<portid>]:proto=<protocol>[:toport=<portid>[-<portid>]][:toaddr=<address>[/<mask>]]
Remove the IPv4 forward port [Z] [O]
Options to Adapt and Query Zones
--add-icmp-block-inversion
Enable inversion of icmp blocks for a zone [Z]
--remove-icmp-block-inversion
Disable inversion of icmp blocks for a zone [Z]
--query-icmp-block-inversion
Return whether inversion of icmp blocks has been enabled
for a zone [Z]
--add-forward Enable forwarding of packets between interfaces and
sources in a zone [Z]
--remove-forward Disable forwarding of packets between interfaces and
sources in a zone [Z]
--query-forward Return whether forwarding of packets between interfaces
and sources has been enabled for a zone [Z]
--get-priority Get the priority [Z]
--set-priority=<priority>
Set the priority [Z]
--get-ingress-priority
Get the ingress priority [Z]
--set-ingress-priority=<priority>
Set the ingress priority [Z]
--get-egress-priority
Get the egress priority [Z]
--set-egress-priority=<priority>
Set the egress priority [Z]
Options to Adapt and Query Policies
--get-priority Get the priority [O]
--set-priority=<priority>
Set the priority [O]
--list-ingress-zones
List ingress zones that are bound to a policy [O]
--add-ingress-zone=<zone>
Add the ingress zone to a policy [O]
--remove-ingress-zone=<zone>
Remove the ingress zone from a policy [O]
--query-ingress-zone=<zone>
Query whether the ingress zone has been adedd to a
policy [O]
--list-egress-zones
List egress zones that are bound to a policy [O]
--add-egress-zone=<zone>
Add the egress zone to a policy [O]
--remove-egress-zone=<zone>
Remove the egress zone from a policy [O]
--query-egress-zone=<zone>
Query whether the egress zone has been adedd to a
policy [O]
Options to Handle Bindings of Interfaces
--list-interfaces List interfaces that are bound to a zone [Z]
--add-interface=<interface>
Bind the <interface> to a zone [Z]
--change-interface=<interface>
Change zone the <interface> is bound to [Z]
--query-interface=<interface>
Query whether <interface> is bound to a zone [Z]
--remove-interface=<interface>
Remove binding of <interface> from a zone [Z]
Options to Handle Bindings of Sources
--list-sources List sources that are bound to a zone [Z]
--add-source=<source>[/<mask>]|<MAC>|ipset:<ipset>
Bind the source to a zone [Z]
--change-source=<source>[/<mask>]|<MAC>|ipset:<ipset>
Change zone the source is bound to [Z]
--query-source=<source>[/<mask>]|<MAC>|ipset:<ipset>
Query whether the source is bound to a zone [Z]
--remove-source=<source>[/<mask>]|<MAC>|ipset:<ipset>
Remove binding of the source from a zone [Z]
Helper Options
--new-helper=<helper> --module=<module> [--family=<family>]
Add a new helper
--new-helper-from-file=<filename> [--name=<helper>]
Add a new helper from file with optional name
--delete-helper=<helper>
Delete an existing helper
--load-helper-defaults=<helper>
Load helper default settings
--info-helper=<helper> Print information about an helper
--path-helper=<helper> Print file path of an helper
--get-helpers Print predefined helpers
--helper=<helper> --set-description=<description>
Set new description to helper
--helper=<helper> --get-description
Print description for helper
--helper=<helper> --set-short=<description>
Set new short description to helper
--helper=<helper> --get-short
Print short description for helper
--helper=<helper> --add-port=<portid>[-<portid>]/<protocol>
Add a new port to helper
--helper=<helper> --remove-port=<portid>[-<portid>]/<protocol>
Remove a port from helper
--helper=<helper> --query-port=<portid>[-<portid>]/<protocol>
Return whether the port has been added for helper
--helper=<helper> --get-ports
List ports of helper
--helper=<helper> --set-module=<module>
Set module to helper
--helper=<helper> --get-module
Get module from helper
--helper=<helper> --set-family={ipv4|ipv6|}
Set family for helper
--helper=<helper> --get-family
Get module from helper
Direct Options
--direct First option for all direct options
--get-all-chains
Get all chains
--get-chains {ipv4|ipv6|eb} <table>
Get all chains added to the table
--add-chain {ipv4|ipv6|eb} <table> <chain>
Add a new chain to the table
--remove-chain {ipv4|ipv6|eb} <table> <chain>
Remove the chain from the table
--query-chain {ipv4|ipv6|eb} <table> <chain>
Return whether the chain has been added to the table
--get-all-rules
Get all rules
--get-rules {ipv4|ipv6|eb} <table> <chain>
Get all rules added to chain in table
--add-rule {ipv4|ipv6|eb} <table> <chain> <priority> <arg>...
Add rule to chain in table
--remove-rule {ipv4|ipv6|eb} <table> <chain> <priority> <arg>...
Remove rule with priority from chain in table
--remove-rules {ipv4|ipv6|eb} <table> <chain>
Remove rules from chain in table
--query-rule {ipv4|ipv6|eb} <table> <chain> <priority> <arg>...
Return whether a rule with priority has been added to
chain in table
--get-all-passthroughs
Get all passthrough rules
--get-passthroughs {ipv4|ipv6|eb} <arg>...
Get passthrough rules
--add-passthrough {ipv4|ipv6|eb} <arg>...
Add a new passthrough rule
--remove-passthrough {ipv4|ipv6|eb} <arg>...
Remove a passthrough rule
--query-passthrough {ipv4|ipv6|eb} <arg>...
Return whether the passthrough rule has been added
Lockdown Options
--lockdown-on Enable lockdown.
--lockdown-off Disable lockdown.
--query-lockdown Query whether lockdown is enabled
Lockdown Whitelist Options
--list-lockdown-whitelist-commands
List all command lines that are on the whitelist
--add-lockdown-whitelist-command=<command>
Add the command to the whitelist
--remove-lockdown-whitelist-command=<command>
Remove the command from the whitelist
--query-lockdown-whitelist-command=<command>
Query whether the command is on the whitelist
--list-lockdown-whitelist-contexts
List all contexts that are on the whitelist
--add-lockdown-whitelist-context=<context>
Add the context context to the whitelist
--remove-lockdown-whitelist-context=<context>
Remove the context from the whitelist
--query-lockdown-whitelist-context=<context>
Query whether the context is on the whitelist
--list-lockdown-whitelist-uids
List all user ids that are on the whitelist
--add-lockdown-whitelist-uid=<uid>
Add the user id uid to the whitelist
--remove-lockdown-whitelist-uid=<uid>
Remove the user id uid from the whitelist
--query-lockdown-whitelist-uid=<uid>
Query whether the user id uid is on the whitelist
--list-lockdown-whitelist-users
List all user names that are on the whitelist
--add-lockdown-whitelist-user=<user>
Add the user name user to the whitelist
--remove-lockdown-whitelist-user=<user>
Remove the user name user from the whitelist
--query-lockdown-whitelist-user=<user>
Query whether the user name user is on the whitelist
Polkit Options
--policy-server Change Polkit actions to 'server' (more restricted)
--policy-desktop Change Polkit actions to 'desktop' (less restricted)
"""
% (SYSTEM_CONFIG_FIREWALL, config.SYSCONFIGDIR)
)
def parse_port_lokkit(value):
try:
(port, proto) = value.split(":")
except Exception:
cmd.fail(
"bad port (most likely missing protocol), correct syntax is portid[-portid]:protocol"
)
return (port, proto)
def pk_symlink(product="server"):
_PK_DIR = "/usr/share/polkit-1/actions/"
_PK_NAME = "org.fedoraproject.FirewallD1."
os.chdir(_PK_DIR)
if os.path.isfile(_PK_NAME + product + ".policy.choice"):
if os.path.isfile(_PK_NAME + "policy"):
os.remove(_PK_NAME + "policy")
os.symlink(_PK_NAME + product + ".policy.choice", _PK_NAME + "policy")
cmd.print_and_exit(
"symlink "
+ _PK_DIR
+ _PK_NAME
+ product
+ ".policy.choice -> "
+ _PK_NAME
+ "policy"
)
else:
cmd.fail("no such file " + _PK_DIR + _PK_NAME + product + ".policy.choice")
# system-config-firewall
def read_sysconfig_args(config_file=SYSTEM_CONFIG_FIREWALL):
filename = None
if os.path.exists(config_file) and os.path.isfile(config_file):
filename = config_file
try:
f = open(filename, "r")
except Exception:
return None
argv = []
for line in f:
if not line:
break
line = line.strip()
if len(line) < 1 or line[0] == "#":
continue
argv.append(line)
f.close()
return argv
parser = argparse.ArgumentParser(
usage="see firewall-offline-cmd man page", add_help=False
)
parser_group_output = parser.add_mutually_exclusive_group()
parser_group_output.add_argument("-v", "--verbose", action="store_true")
parser_group_output.add_argument("-q", "--quiet", action="store_true")
parser_group_lokkit = parser.add_argument_group()
parser_group_lokkit.add_argument("--enabled", action="store_true")
parser_group_lokkit.add_argument("--disabled", action="store_true")
parser_group_lokkit.add_argument("--addmodule", metavar="<module>", action="append")
parser_group_lokkit.add_argument("--removemodule", metavar="<module>", action="append")
parser_group_lokkit.add_argument(
"--service", "-s", metavar="<service>", action="append"
)
parser_group_lokkit.add_argument(
"--remove-service", metavar="<service>", action="append"
)
parser_group_lokkit.add_argument("--port", "-p", metavar="<port>", action="append")
parser_group_lokkit.add_argument("--trust", "-t", metavar="<iface>", action="append")
parser_group_lokkit.add_argument("--masq", "-m", metavar="<iface>", action="append")
parser_group_lokkit.add_argument(
"--custom-rules", metavar="<filename>", action="append"
)
parser_group_lokkit.add_argument("--forward-port", metavar="<port>", action="append")
parser_group_lokkit.add_argument("--block-icmp", metavar="<icmptype>", action="append")
parser.add_argument("--system-config", metavar="path")
parser.add_argument("--default-config", metavar="path")
parser.add_argument("--check-config", action="store_true")
parser_group_standalone = parser.add_mutually_exclusive_group()
parser_group_standalone.add_argument("-h", "--help", action="store_true")
parser_group_standalone.add_argument("-V", "--version", action="store_true")
parser_group_standalone.add_argument("--get-log-denied", action="store_true")
parser_group_standalone.add_argument("--set-log-denied", metavar="<value>")
parser_group_standalone.add_argument("--get-automatic-helpers", action="store_true")
parser_group_standalone.add_argument("--set-automatic-helpers", metavar="<value>")
parser_group_standalone.add_argument("--policy-server", action="store_true")
parser_group_standalone.add_argument("--policy-desktop", action="store_true")
parser_group_standalone.add_argument("--lockdown-on", action="store_true")
parser_group_standalone.add_argument("--lockdown-off", action="store_true")
parser_group_standalone.add_argument("--query-lockdown", action="store_true")
parser_group_standalone.add_argument("--reset-to-defaults", action="store_true")
parser_group_standalone.add_argument("--get-default-zone", action="store_true")
parser_group_standalone.add_argument("--set-default-zone", metavar="<zone>")
parser_group_standalone.add_argument("--get-zones", action="store_true")
parser_group_standalone.add_argument("--get-policies", action="store_true")
parser_group_standalone.add_argument("--get-services", action="store_true")
parser_group_standalone.add_argument("--get-icmptypes", action="store_true")
parser_group_standalone.add_argument(
"--get-zone-of-interface", metavar="<iface>", action="append"
)
parser_group_standalone.add_argument(
"--get-zone-of-source", metavar="<source>", action="append"
)
parser_group_standalone.add_argument("--list-all-zones", action="store_true")
parser_group_standalone.add_argument("--list-all-policies", action="store_true")
parser_group_standalone.add_argument("--info-zone", metavar="<zone>")
parser_group_standalone.add_argument("--info-policy", metavar="<policy>")
parser_group_standalone.add_argument("--info-service", metavar="<service>")
parser_group_standalone.add_argument("--info-icmptype", metavar="<icmptype>")
parser_group_standalone.add_argument("--info-ipset", metavar="<ipset>")
parser_group_standalone.add_argument("--info-helper", metavar="<helper>")
parser_group_config = parser.add_mutually_exclusive_group()
parser_group_config.add_argument("--new-icmptype", metavar="<icmptype>")
parser_group_config.add_argument("--new-icmptype-from-file", metavar="<filename>")
parser_group_config.add_argument("--delete-icmptype", metavar="<icmptype>")
parser_group_config.add_argument("--load-icmptype-defaults", metavar="<icmptype>")
parser_group_config.add_argument("--new-service", metavar="<service>")
parser_group_config.add_argument("--new-service-from-file", metavar="<filename>")
parser_group_config.add_argument("--delete-service", metavar="<service>")
parser_group_config.add_argument("--load-service-defaults", metavar="<service>")
parser_group_config.add_argument("--new-zone", metavar="<zone>")
parser_group_config.add_argument("--new-zone-from-file", metavar="<filename>")
parser_group_config.add_argument("--delete-zone", metavar="<zone>")
parser_group_config.add_argument("--load-zone-defaults", metavar="<zone>")
parser_group_config.add_argument("--new-policy", metavar="<policy>")
parser_group_config.add_argument("--new-policy-from-file", metavar="<filename>")
parser_group_config.add_argument("--delete-policy", metavar="<policy>")
parser_group_config.add_argument("--load-policy-defaults", metavar="<policy>")
parser_group_config.add_argument("--new-ipset", metavar="<ipset>")
parser_group_config.add_argument("--new-ipset-from-file", metavar="<filename>")
parser_group_config.add_argument("--delete-ipset", metavar="<ipset>")
parser_group_config.add_argument("--load-ipset-defaults", metavar="<ipset>")
parser_group_config.add_argument("--new-helper", metavar="<helper>")
parser_group_config.add_argument("--new-helper-from-file", metavar="<filename>")
parser_group_config.add_argument("--delete-helper", metavar="<helper>")
parser_group_config.add_argument("--load-helper-defaults", metavar="<helper>")
parser_group_config.add_argument("--path-zone", metavar="<zone>")
parser_group_config.add_argument("--path-policy", metavar="<policy>")
parser_group_config.add_argument("--path-service", metavar="<service>")
parser_group_config.add_argument("--path-icmptype", metavar="<icmptype>")
parser_group_config.add_argument("--path-ipset", metavar="<ipset>")
parser_group_config.add_argument("--path-helper", metavar="<helper>")
parser.add_argument("--name", default="", metavar="<name>")
parser_group_lockdown_whitelist = parser.add_mutually_exclusive_group()
parser_group_lockdown_whitelist.add_argument(
"--list-lockdown-whitelist-commands", action="store_true"
)
parser_group_lockdown_whitelist.add_argument(
"--add-lockdown-whitelist-command", metavar="<command>", action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--remove-lockdown-whitelist-command", metavar="<command>", action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--query-lockdown-whitelist-command", metavar="<command>", action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--list-lockdown-whitelist-contexts", action="store_true"
)
parser_group_lockdown_whitelist.add_argument(
"--add-lockdown-whitelist-context", metavar="<context>", action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--remove-lockdown-whitelist-context", metavar="<context>", action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--query-lockdown-whitelist-context", metavar="<context>", action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--list-lockdown-whitelist-uids", action="store_true"
)
parser_group_lockdown_whitelist.add_argument(
"--add-lockdown-whitelist-uid", metavar="<uid>", type=int, action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--remove-lockdown-whitelist-uid", metavar="<uid>", type=int, action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--query-lockdown-whitelist-uid", metavar="<uid>", type=int, action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--list-lockdown-whitelist-users", action="store_true"
)
parser_group_lockdown_whitelist.add_argument(
"--add-lockdown-whitelist-user", metavar="<user>", action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--remove-lockdown-whitelist-user", metavar="<user>", action="append"
)
parser_group_lockdown_whitelist.add_argument(
"--query-lockdown-whitelist-user", metavar="<user>", action="append"
)
parser.add_argument("--zone", default="", metavar="<zone>")
parser.add_argument("--policy", default="", metavar="<policy>")
parser_group_zone_or_policy = parser.add_mutually_exclusive_group()
parser_group_zone_or_policy.add_argument(
"--add-interface", metavar="<iface>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-interface", metavar="<iface>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-interface", metavar="<iface>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--change-interface", "--change-zone", metavar="<iface>", action="append"
)
parser_group_zone_or_policy.add_argument("--list-interfaces", action="store_true")
parser_group_zone_or_policy.add_argument(
"--add-source", metavar="<source>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-source", metavar="<source>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-source", metavar="<source>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--change-source", metavar="<source>", action="append"
)
parser_group_zone_or_policy.add_argument("--list-sources", action="store_true")
parser_group_zone_or_policy.add_argument(
"--add-ingress-zone", metavar="<zone>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-ingress-zone", metavar="<zone>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-ingress-zone", metavar="<zone>", action="append"
)
parser_group_zone_or_policy.add_argument("--list-ingress-zones", action="store_true")
parser_group_zone_or_policy.add_argument(
"--add-egress-zone", metavar="<zone>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-egress-zone", metavar="<zone>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-egress-zone", metavar="<zone>", action="append"
)
parser_group_zone_or_policy.add_argument("--list-egress-zones", action="store_true")
parser_group_zone_or_policy.add_argument(
"--add-rich-rule", metavar="<rule>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-rich-rule", metavar="<rule>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-rich-rule", metavar="<rule>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--add-service", metavar="<service>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-service-from-zone", metavar="<service>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-service-from-policy", metavar="<service>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-service", metavar="<service>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--add-port", metavar="<port>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-port", metavar="<port>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-port", metavar="<port>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--add-protocol", metavar="<protocol>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-protocol", metavar="<protocol>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-protocol", metavar="<protocol>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--add-source-port", metavar="<port>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-source-port", metavar="<port>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-source-port", metavar="<port>", action="append"
)
parser_group_zone_or_policy.add_argument("--add-forward", action="store_true")
parser_group_zone_or_policy.add_argument("--remove-forward", action="store_true")
parser_group_zone_or_policy.add_argument("--query-forward", action="store_true")
parser_group_zone_or_policy.add_argument("--add-masquerade", action="store_true")
parser_group_zone_or_policy.add_argument("--remove-masquerade", action="store_true")
parser_group_zone_or_policy.add_argument("--query-masquerade", action="store_true")
parser_group_zone_or_policy.add_argument(
"--add-icmp-block", metavar="<icmptype>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-icmp-block", metavar="<icmptype>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-icmp-block", metavar="<icmptype>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--add-icmp-block-inversion", action="store_true"
)
parser_group_zone_or_policy.add_argument(
"--remove-icmp-block-inversion", action="store_true"
)
parser_group_zone_or_policy.add_argument(
"--query-icmp-block-inversion", action="store_true"
)
parser_group_zone_or_policy.add_argument(
"--add-forward-port", metavar="<port>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--remove-forward-port", metavar="<port>", action="append"
)
parser_group_zone_or_policy.add_argument(
"--query-forward-port", metavar="<port>", action="append"
)
parser_group_zone_or_policy.add_argument("--list-rich-rules", action="store_true")
parser_group_zone_or_policy.add_argument("--list-services", action="store_true")
parser_group_zone_or_policy.add_argument("--list-ports", action="store_true")
parser_group_zone_or_policy.add_argument("--list-protocols", action="store_true")
parser_group_zone_or_policy.add_argument("--list-icmp-blocks", action="store_true")
parser_group_zone_or_policy.add_argument("--list-forward-ports", action="store_true")
parser_group_zone_or_policy.add_argument("--list-source-ports", action="store_true")
parser_group_zone_or_policy.add_argument("--list-all", action="store_true")
parser_group_zone_or_policy.add_argument("--get-target", action="store_true")
parser_group_zone_or_policy.add_argument("--set-target", metavar="<target>")
parser_group_zone_or_policy.add_argument("--get-priority", action="store_true")
parser_group_zone_or_policy.add_argument("--set-priority", metavar="<priority>")
parser_group_zone_or_policy.add_argument("--get-ingress-priority", action="store_true")
parser_group_zone_or_policy.add_argument("--set-ingress-priority", metavar="<priority>")
parser_group_zone_or_policy.add_argument("--get-egress-priority", action="store_true")
parser_group_zone_or_policy.add_argument("--set-egress-priority", metavar="<priority>")
parser.add_argument("--option", metavar="<key>[=<value>]", action="append")
parser.add_argument("--type", metavar="<ipsettype>")
parser.add_argument("--ipset", metavar="<ipset>")
parser_ipset = parser.add_mutually_exclusive_group()
# parser_ipset.add_argument("--add-option", metavar="<key>[=<value>]")
# parser_ipset.add_argument("--remove-option", metavar="<key>[=<value>]")
# parser_ipset.add_argument("--query-option", metavar="<key>[=<value>]")
# parser_ipset.add_argument("--get-options", action="store_true")
parser_ipset.add_argument("--get-ipsets", action="store_true")
parser_ipset.add_argument("--add-entry", metavar="<entry>", action="append")
parser_ipset.add_argument("--remove-entry", metavar="<entry>", action="append")
parser_ipset.add_argument("--query-entry", metavar="<entry>", action="append")
parser_ipset.add_argument("--get-entries", action="store_true")
parser_ipset.add_argument(
"--add-entries-from-file", metavar="<filename>", action="append"
)
parser_ipset.add_argument(
"--remove-entries-from-file", metavar="<filename>", action="append"
)
parser.add_argument("--icmptype", metavar="<icmptype>")
parser_icmptype = parser.add_mutually_exclusive_group()
parser_icmptype.add_argument("--add-destination", metavar="<ipv>", action="append")
parser_icmptype.add_argument("--remove-destination", metavar="<ipv>", action="append")
parser_icmptype.add_argument("--query-destination", metavar="<ipv>", action="append")
parser_icmptype.add_argument("--get-destinations", action="store_true")
parser_service = parser.add_mutually_exclusive_group()
parser_service.add_argument("--get-ports", action="store_true")
parser_service.add_argument("--get-source-ports", action="store_true")
parser_service.add_argument("--get-protocols", action="store_true")
parser_service.add_argument("--add-module", metavar="<module>", action="append")
parser_service.add_argument("--remove-module", metavar="<module>", action="append")
parser_service.add_argument("--query-module", metavar="<module>", action="append")
parser_service.add_argument("--get-modules", action="store_true")
parser_service.add_argument("--add-helper", metavar="<helper>", action="append")
parser_service.add_argument("--remove-helper", metavar="<helper>", action="append")
parser_service.add_argument("--query-helper", metavar="<helper>", action="append")
parser_service.add_argument("--get-service-helpers", action="store_true")
parser_service.add_argument("--add-include", metavar="<service>", action="append")
parser_service.add_argument("--remove-include", metavar="<service>", action="append")
parser_service.add_argument("--query-include", metavar="<service>", action="append")
parser_service.add_argument("--get-includes", action="store_true")
parser_service.add_argument(
"--set-destination", metavar="<destination>", action="append"
)
parser_service.add_argument("--get-destination", action="store_true")
parser_service.add_argument("--set-description", metavar="<description>")
parser_service.add_argument("--get-description", action="store_true")
parser_service.add_argument("--set-short", metavar="<description>")
parser_service.add_argument("--get-short", action="store_true")
parser.add_argument("--helper", metavar="<helper>")
parser.add_argument("--family", metavar="<family>")
parser.add_argument("--module", metavar="<module>")
parser_helper = parser.add_mutually_exclusive_group()
# parser_helper.add_argument("--get-ports", action="store_true")
parser_helper.add_argument("--get-helpers", action="store_true")
parser_helper.add_argument("--set-module", metavar="<module>")
parser_helper.add_argument("--get-module", action="store_true")
# parser_helper.add_argument("--query-module", metavar="<module>")
parser_helper.add_argument("--set-family", metavar="<family>|''", nargs="*")
parser_helper.add_argument("--get-family", action="store_true")
parser.add_argument("--direct", action="store_true")
# not possible to have sequences of options here
parser_direct = parser.add_mutually_exclusive_group()
parser_direct.add_argument(
"--add-passthrough",
nargs=argparse.REMAINDER,
metavar=("{ ipv4 | ipv6 | eb }", "<args>"),
)
parser_direct.add_argument(
"--remove-passthrough",
nargs=argparse.REMAINDER,
metavar=("{ ipv4 | ipv6 | eb }", "<args>"),
)
parser_direct.add_argument(
"--query-passthrough",
nargs=argparse.REMAINDER,
metavar=("{ ipv4 | ipv6 | eb }", "<args>"),
)
parser_direct.add_argument(
"--get-passthroughs", nargs=1, metavar=("{ ipv4 | ipv6 | eb }")
)
parser_direct.add_argument("--get-all-passthroughs", action="store_true")
parser_direct.add_argument(
"--add-chain", nargs=3, metavar=("{ ipv4 | ipv6 | eb }", "<table>", "<chain>")
)
parser_direct.add_argument(
"--remove-chain", nargs=3, metavar=("{ ipv4 | ipv6 | eb }", "<table>", "<chain>")
)
parser_direct.add_argument(
"--query-chain", nargs=3, metavar=("{ ipv4 | ipv6 | eb }", "<table>", "<chain>")
)
parser_direct.add_argument("--get-all-chains", action="store_true")
parser_direct.add_argument(
"--get-chains", nargs=2, metavar=("{ ipv4 | ipv6 | eb }", "<table>")
)
parser_direct.add_argument(
"--add-rule",
nargs=argparse.REMAINDER,
metavar=("{ ipv4 | ipv6 | eb }", "<table> <chain> <priority> <args>"),
)
parser_direct.add_argument(
"--remove-rule",
nargs=argparse.REMAINDER,
metavar=("{ ipv4 | ipv6 | eb }", "<table> <chain> <priority> <args>"),
)
parser_direct.add_argument(
"--remove-rules", nargs=3, metavar=("{ ipv4 | ipv6 | eb }", "<table> <chain>")
)
parser_direct.add_argument(
"--query-rule",
nargs=argparse.REMAINDER,
metavar=("{ ipv4 | ipv6 | eb }", "<table> <chain> <priority> <args>"),
)
parser_direct.add_argument(
"--get-rules", nargs=3, metavar=("{ ipv4 | ipv6 | eb }", "<table>", "<chain>")
)
parser_direct.add_argument("--get-all-rules", action="store_true")
##############################################################################
cmd = FirewallCommand()
def myexcepthook(exctype, value, traceback):
cmd.exception_handler(str(value))
sys.excepthook = myexcepthook
if len(sys.argv) > 1 and any(
"--migrate-system-config-firewall" in arg for arg in sys.argv
):
args = sys.argv[1:]
migration_parser = argparse.ArgumentParser(
usage="see firewall-offline-cmd man page", add_help=False
)
migration_parser.add_argument("-h", "--help", action="store_true")
migration_parser.add_argument("-v", "--verbose", action="store_true")
migration_parser.add_argument("-q", "--quiet", action="store_true")
migration_parser.add_argument(
"--migrate-system-config-firewall", metavar="<file>", action="store"
)
a, unknown = migration_parser.parse_known_args(args)
cmd.set_quiet(a.quiet)
cmd.set_verbose(a.verbose)
if a.help:
__usage()
sys.exit(0)
else:
assert_root()
if a.quiet:
# it makes no sense to use --quiet with these options
a.quiet = False
cmd.set_quiet(a.quiet)
cmd.fail("-q/--quiet can't be used with this option(s)")
if a.migrate_system_config_firewall:
args = read_sysconfig_args(a.migrate_system_config_firewall)
if not args:
cmd.fail(
"Opening of '%s' failed, exiting." % a.migrate_system_config_firewall
)
args += unknown
elif len(sys.argv) > 1:
i = -1
args = sys.argv[1:]
if "--add-passthrough" in args:
i = args.index("--add-passthrough") + 1
elif "--remove-passthrough" in args:
i = args.index("--remove-passthrough") + 1
elif "--query-passthrough" in args:
i = args.index("--query-passthrough") + 1
elif "--add-rule" in args:
i = args.index("--add-rule") + 4
elif "--remove-rule" in args:
i = args.index("--remove-rule") + 4
elif "--query-rule" in args:
i = args.index("--query-rule") + 4
# join <args> into one argument to prevent parser from parsing each iptables
# option, because they can conflict with firewall-cmd options
# # e.g. --delete (iptables) and --delete-* (firewall-cmd)
if (i > -1) and (i < len(args) - 1):
aux_args = args[:]
args = aux_args[: i + 1] # all but not <args>
args.append(joinArgs(aux_args[i + 1 :])) # add <args> as one arg
else:
assert_root()
# migrate configuration from SYSTEM_CONFIG_FIREWALL
args = read_sysconfig_args()
if not args:
cmd.fail("Opening of '%s' failed, exiting." % SYSTEM_CONFIG_FIREWALL)
a = parser.parse_args(args)
options_lokkit = (
a.enabled
or a.disabled
or a.addmodule
or a.removemodule
or a.trust
or a.masq
or a.custom_rules
or a.service
or a.remove_service
or a.port
or a.trust
or a.masq
or a.forward_port
or a.block_icmp
)
options_standalone = (
a.help
or a.version
or a.policy_server
or a.policy_desktop
or a.lockdown_on
or a.lockdown_off
or a.query_lockdown
or a.reset_to_defaults
or a.get_default_zone
or a.set_default_zone
or a.get_log_denied
or a.set_log_denied
or a.get_automatic_helpers
or a.set_automatic_helpers
)
options_desc_xml_file = (
a.set_description or a.get_description or a.set_short or a.get_short
)
options_lockdown_whitelist = (
a.list_lockdown_whitelist_commands
or a.add_lockdown_whitelist_command
or a.remove_lockdown_whitelist_command
or a.query_lockdown_whitelist_command
or a.list_lockdown_whitelist_contexts
or a.add_lockdown_whitelist_context
or a.remove_lockdown_whitelist_context
or a.query_lockdown_whitelist_context
or a.list_lockdown_whitelist_uids
or a.add_lockdown_whitelist_uid is not None
or a.remove_lockdown_whitelist_uid is not None
or a.query_lockdown_whitelist_uid is not None
or a.list_lockdown_whitelist_users
or a.add_lockdown_whitelist_user
or a.remove_lockdown_whitelist_user
or a.query_lockdown_whitelist_user
)
options_config = (
a.get_zones
or a.get_services
or a.get_icmptypes
or options_lockdown_whitelist
or a.list_all_zones
or a.get_zone_of_interface
or a.get_zone_of_source
or a.info_zone
or a.info_icmptype
or a.info_service
or a.info_ipset
or a.info_policy
or a.get_ipsets
or a.info_helper
or a.get_helpers
or a.get_policies
or a.list_all_policies
)
options_zone_and_policy_adapt_query = (
a.add_service
or a.remove_service_from_zone
or a.query_service
or a.add_port
or a.remove_port
or a.query_port
or a.add_protocol
or a.remove_protocol
or a.query_protocol
or a.add_source_port
or a.remove_source_port
or a.query_source_port
or a.add_icmp_block
or a.remove_icmp_block
or a.query_icmp_block
or a.add_forward_port
or a.remove_forward_port
or a.query_forward_port
or a.add_rich_rule
or a.remove_rich_rule
or a.query_rich_rule
or a.add_masquerade
or a.remove_masquerade
or a.query_masquerade
or a.list_services
or a.list_ports
or a.list_protocols
or a.list_source_ports
or a.list_icmp_blocks
or a.list_forward_ports
or a.list_rich_rules
or a.list_all
or a.get_target
or a.set_target
or a.set_priority
or a.get_priority
)
options_zone_unique = (
a.add_icmp_block_inversion
or a.remove_icmp_block_inversion
or a.query_icmp_block_inversion
or a.add_forward
or a.remove_forward
or a.query_forward
or a.list_interfaces
or a.change_interface
or a.add_interface
or a.remove_interface
or a.query_interface
or a.list_sources
or a.change_source
or a.add_source
or a.remove_source
or a.query_source
or a.set_ingress_priority
or a.get_ingress_priority
or a.set_egress_priority
or a.get_egress_priority
)
options_zone_ops = options_zone_unique or options_zone_and_policy_adapt_query
options_policy_unique = (
a.list_ingress_zones
or a.add_ingress_zone
or a.remove_ingress_zone
or a.query_ingress_zone
or a.list_egress_zones
or a.add_egress_zone
or a.remove_egress_zone
or a.query_egress_zone
)
options_policy_ops = options_policy_unique or options_zone_and_policy_adapt_query
options_zone = a.zone or options_zone_ops or options_desc_xml_file
options_policy = a.policy or options_policy_ops or options_desc_xml_file
options_ipset = (
a.add_entry
or a.remove_entry
or a.query_entry
or a.get_entries
or a.add_entries_from_file
or a.remove_entries_from_file
or options_desc_xml_file
)
options_icmptype = (
a.add_destination
or a.remove_destination
or a.query_destination
or a.get_destinations
or options_desc_xml_file
)
options_service = (
a.add_port
or a.remove_port
or a.query_port
or a.get_ports
or a.add_protocol
or a.remove_protocol
or a.query_protocol
or a.get_protocols
or a.add_source_port
or a.remove_source_port
or a.query_source_port
or a.get_source_ports
or a.add_module
or a.remove_module
or a.query_module
or a.get_modules
or a.set_destination
or a.remove_destination
or a.query_destination
or a.get_destinations
or options_desc_xml_file
or a.add_include
or a.remove_include
or a.query_include
or a.get_includes
or a.add_helper
or a.remove_helper
or a.query_helper
or a.get_service_helpers
)
options_helper = (
a.add_port
or a.remove_port
or a.query_port
or a.get_ports
or a.set_module
or a.get_module
or a.set_family
or a.get_family
or options_desc_xml_file
)
options_permanent = (
options_config
or options_zone
or options_policy
or a.new_icmptype
or a.delete_icmptype
or a.new_icmptype_from_file
or a.load_icmptype_defaults
or a.new_service
or a.delete_service
or a.new_service_from_file
or a.load_service_defaults
or a.new_zone
or a.delete_zone
or a.new_zone_from_file
or a.load_zone_defaults
or a.new_policy
or a.delete_policy
or a.new_policy_from_file
or a.load_policy_defaults
or a.new_helper
or a.delete_helper
or a.new_helper_from_file
or a.load_helper_defaults
or a.new_ipset
or a.delete_ipset
or a.new_ipset_from_file
or a.load_ipset_defaults
or a.ipset
or options_ipset
or (a.icmptype and options_icmptype)
or (a.service and options_service)
or (a.helper and options_helper)
or a.path_zone
or a.path_icmptype
or a.path_service
or a.path_ipset
or a.path_helper
or a.path_policy
)
options_direct = (
a.add_chain
or a.remove_chain
or a.query_chain
or a.get_chains
or a.get_all_chains
or a.add_rule
or a.remove_rule
or a.remove_rules
or a.query_rule
or a.get_rules
or a.get_all_rules
or a.add_passthrough
or a.remove_passthrough
or a.query_passthrough
or a.get_passthroughs
or a.get_all_passthroughs
)
# these are supposed to only write out some output
options_list_get = (
a.help
or a.version
or a.list_all
or a.list_all_zones
or a.list_lockdown_whitelist_commands
or a.list_lockdown_whitelist_contexts
or a.list_lockdown_whitelist_uids
or a.list_lockdown_whitelist_users
or a.list_services
or a.list_ports
or a.list_protocols
or a.list_icmp_blocks
or a.list_forward_ports
or a.list_rich_rules
or a.list_interfaces
or a.list_sources
or a.get_default_zone
or a.get_zone_of_interface
or a.get_zone_of_source
or a.get_zones
or a.get_services
or a.get_icmptypes
or a.get_target
or a.info_zone
or a.info_icmptype
or a.info_service
or a.info_ipset
or a.get_ipsets
or a.get_entries
or a.info_helper
or a.get_helpers
or a.get_destinations
or a.get_description
or a.list_all_policies
or a.info_policy
or a.get_policies
)
# Set quiet and verbose
cmd.set_quiet(a.quiet)
cmd.set_verbose(a.verbose)
# Check various impossible combinations of options
if not (
options_standalone
or options_ipset
or options_lokkit
or options_icmptype
or options_service
or options_helper
or options_permanent
or options_direct
or options_desc_xml_file
or a.check_config
):
cmd.fail(parser.format_usage() + "No option specified.")
if (
options_lokkit
and (options_standalone or options_permanent or options_direct)
and not (options_service and a.service)
):
cmd.fail(parser.format_usage() + "Can't use lokkit options with other options.")
if options_standalone and (options_permanent or options_direct or options_ipset):
cmd.fail(
parser.format_usage() + "Can't use stand-alone options with other options."
)
if options_ipset and not options_desc_xml_file and not a.ipset:
cmd.fail(parser.format_usage() + "No ipset specified.")
if (
(options_icmptype and not a.icmptype)
and not (options_service and a.service)
and not options_desc_xml_file
):
cmd.fail(parser.format_usage() + "No icmptype specified.")
if options_service and a.service and len(a.service) > 0:
if len(a.service) > 1:
cmd.fail(parser.format_usage() + "More than one service specified.")
# use the first entry in the array only
a.service = a.service[0]
if (
(options_helper and not a.helper)
and not (options_service and a.service)
and not options_zone
and not options_desc_xml_file
and not options_policy
):
cmd.fail(parser.format_usage() + "No helper specified.")
if options_direct and (options_zone or options_policy):
cmd.fail(parser.format_usage() + "Can't use 'direct' options with other options.")
if (a.direct and not options_direct) or (options_direct and not a.direct):
cmd.fail(parser.format_usage() + "Wrong usage of 'direct' options.")
if a.name and not (
a.new_zone_from_file
or a.new_service_from_file
or a.new_ipset_from_file
or a.new_icmptype_from_file
or a.new_helper_from_file
or a.new_policy_from_file
):
cmd.fail(parser.format_usage() + "Wrong usage of '--name' option.")
if options_config and (options_zone or options_policy):
cmd.fail(
parser.format_usage()
+ "Wrong usage of --get-zones | --get-services | --get-icmptypes | --get-policies."
)
if a.quiet and options_list_get:
# it makes no sense to use --quiet with these options
a.quiet = False
cmd.set_quiet(a.quiet)
cmd.fail("-q/--quiet can't be used with this option(s)")
if a.zone and a.policy:
cmd.fail(parser.format_usage() + "Can't use --zone with --policy.")
if a.policy and options_zone_unique:
cmd.fail(parser.format_usage() + "Can't use --policy with zone only options.")
if a.zone and options_policy_unique:
cmd.fail(parser.format_usage() + "Can't use --zone with policy only options.")
if not a.policy and options_policy_unique:
cmd.fail(parser.format_usage() + "Must use --policy with policy only options.")
if a.help:
__usage()
sys.exit(0)
assert_root()
if a.system_config:
config.set_system_config_paths(a.system_config)
if a.default_config:
config.set_default_config_paths(a.default_config)
if a.check_config:
try:
fw = Firewall(offline=True)
fw.start()
check_on_disk_config(fw)
except FirewallError as error:
cmd.print_and_exit("Configuration error: %s" % error, error.code)
except Exception as msg:
cmd.fail("Configuration error: %s" % msg)
sys.exit(0)
zone = a.zone
fw = Firewall(offline=True)
fw.start()
try:
# Lokkit Compatibility Options
if options_lokkit and not (options_service and a.service):
trusted_zone = "trusted"
default_zone = fw.get_default_zone()
fw_zone = fw.config.get_zone(default_zone)
fw_settings = FirewallClientZoneSettings(
fw.config.get_zone_config_dict(fw_zone)
)
if a.enabled:
# Enable firewall (default)
os.system("systemctl enable firewalld.service")
if a.disabled:
# Disable firewall
os.system("systemctl disable firewalld.service")
if a.addmodule:
for m in a.addmodule:
cmd.print_msg("Ignoring addmodule '%s'" % m)
if a.removemodule:
for m in a.removemodule:
cmd.print_msg("Ignoring removemodule '%s'" % m)
if a.custom_rules:
for c in a.custom_rules:
cmd.print_msg("Ignoring custom-rule '%s'" % c)
if a.service:
for s in a.service:
cmd.print_msg("Adding service '%s' to default zone." % s)
if not fw_settings.queryService(s):
fw_settings.addService(s)
else:
cmd.print_msg("ALREADY_ENABLED: %s" % s)
if a.remove_service:
for s in a.remove_service:
cmd.print_msg("Removing service '%s' from default zone." % s)
if fw_settings.queryService(s):
fw_settings.removeService(s)
else:
cmd.print_msg("NOT_ENABLED: %s" % s)
if a.port:
for port_proto in a.port:
(port, proto) = parse_port_lokkit(port_proto)
cmd.print_msg("Adding port '%s/%s' to default zone." % (port, proto))
if not fw_settings.queryPort(port, proto):
fw_settings.addPort(port, proto)
else:
cmd.print_msg("ALREADY_ENABLED: %s" % port_proto)
if a.trust:
if default_zone != trusted_zone:
fw_trusted = fw.config.get_zone("trusted")
fw_trusted_settings = FirewallClientZoneSettings(
fw.config.get_zone_config_dict(fw_trusted)
)
# Bind an interface to the trusted zone
for i in a.trust:
cmd.print_msg(
"Interface '%s' will be bound to zone '%s'." % (i, trusted_zone)
)
if not fw_trusted_settings.queryInterface(i):
fw_trusted_settings.addInterface(i)
else:
cmd.print_msg("ALREADY_ENABLED: %s" % i)
fw.config.set_zone_config_dict(
fw_trusted, fw_trusted_settings.getSettingsDict()
)
else:
for i in a.trust:
cmd.print_msg(
"Interface '%s' will be bound to zone '%s'." % (i, trusted_zone)
)
if not fw_settings.queryInterface(i):
fw_settings.addInterface(i)
else:
cmd.print_msg("ALREADY_ENABLED: %s" % i)
if a.masq:
# Enables masquerading in the default zone, interface argument is ignored
cmd.print_msg("Enabling masquerade for the default zone.")
fw_settings.setMasquerade(True)
if a.forward_port:
for fp in a.forward_port:
(port, protocol, toport, toaddr) = cmd.parse_forward_port(
fp, compat=True
)
cmd.print_msg(
"Adding forward port %s:%s:%s:%s to default zone."
% (port, protocol, toport, toaddr)
)
if not fw_settings.queryForwardPort(port, protocol, toport, toaddr):
fw_settings.addForwardPort(port, protocol, toport, toaddr)
else:
cmd.print_msg("ALREADY_ENABLED: %s" % fp)
if a.block_icmp:
for ib in a.block_icmp:
cmd.print_msg("Adding icmpblock '%s' to default zone." % ib)
if not fw_settings.queryIcmpBlock(ib):
fw_settings.addIcmpBlock(ib)
else:
cmd.print_msg("ALREADY_ENABLED: %s" % ib)
fw.config.set_zone_config_dict(fw_zone, fw_settings.getSettingsDict())
elif a.version:
cmd.print_and_exit(config.VERSION)
elif a.get_log_denied:
cmd.print_and_exit(fw.get_log_denied())
elif a.set_log_denied:
fw.set_log_denied(a.set_log_denied)
elif a.get_automatic_helpers:
cmd.print_and_exit(fw.get_automatic_helpers())
elif a.set_automatic_helpers:
fw.set_automatic_helpers(a.set_automatic_helpers)
elif a.policy_server:
pk_symlink("server")
elif a.policy_desktop:
pk_symlink("desktop")
# options from firewall-cmd
elif a.get_default_zone:
cmd.print_and_exit(fw.get_default_zone())
elif a.set_default_zone:
fw.set_default_zone(a.set_default_zone)
# lockdown
elif a.lockdown_on:
fw.enable_lockdown()
elif a.lockdown_off:
fw.disable_lockdown()
elif a.query_lockdown:
cmd.print_query_result(fw.policies.query_lockdown())
# reset defaults
elif a.reset_to_defaults:
fw.config.reset_defaults()
# zones
elif a.get_zones:
zones = fw.config.get_zones()
cmd.print_and_exit(" ".join(zones))
elif a.new_zone:
fw.config.new_zone_dict(
a.new_zone, FirewallClientZoneSettings().getSettingsDict()
)
elif a.new_zone_from_file:
filename = os.path.basename(a.new_zone_from_file)
dirname = os.path.dirname(a.new_zone_from_file)
if dirname == "":
dirname = "./"
try:
obj = zone_reader(filename, dirname)
except FirewallError as msg:
cmd.print_and_exit(
"Failed to load zone file '%s': %s" % (a.new_zone_from_file, msg),
msg.code,
)
except IOError as msg:
cmd.fail("Failed to load zone file: %s" % msg)
if a.name:
obj.name = a.name
fw.config.new_zone(obj.name, obj.export_config())
elif a.delete_zone:
obj = fw.config.get_zone(a.delete_zone)
fw.config.remove_zone(obj)
elif a.load_zone_defaults:
obj = fw.config.get_zone(a.load_zone_defaults)
fw.config.load_zone_defaults(obj)
elif a.info_zone:
zone = fw.config.get_zone(a.info_zone)
settings = FirewallClientZoneSettings(fw.config.get_zone_config_dict(zone))
cmd.print_zone_info(a.info_zone, settings, default_zone=fw.get_default_zone())
sys.exit(0)
elif a.path_zone:
obj = fw.config.get_zone(a.path_zone)
cmd.print_and_exit("%s/%s" % (obj.path, obj.filename))
# policies
elif a.get_policies:
policies = fw.config.get_policy_objects()
cmd.print_and_exit(" ".join(policies))
elif a.new_policy:
fw.config.new_policy_object_dict(
a.new_policy, FirewallClientPolicySettings().getSettingsDict()
)
elif a.new_policy_from_file:
filename = os.path.basename(a.new_policy_from_file)
dirname = os.path.dirname(a.new_policy_from_file)
if dirname == "":
dirname = "./"
try:
obj = policy_reader(filename, dirname)
except FirewallError as msg:
cmd.print_and_exit(
"Failed to load policy file '%s': %s" % (a.new_policy_from_file, msg),
msg.code,
)
except IOError as msg:
cmd.fail("Failed to load policy file: %s" % msg)
if a.name:
obj.name = a.name
fw.config.new_policy_object_dict(obj.name, obj.export_config_dict())
elif a.delete_policy:
obj = fw.config.get_policy_object(a.delete_policy)
fw.config.remove_policy_object(obj)
elif a.load_policy_defaults:
obj = fw.config.get_policy_object(a.load_policy_defaults)
fw.config.load_policy_object_defaults(obj)
elif a.info_policy:
policy = fw.config.get_policy_object(a.info_policy)
settings = FirewallClientPolicySettings(
fw.config.get_policy_object_config_dict(policy)
)
cmd.print_policy_info(a.info_policy, settings, True)
sys.exit(0)
elif a.path_policy:
obj = fw.config.get_policy_object(a.path_policy)
cmd.print_and_exit("%s/%s" % (obj.path, obj.filename))
# services
elif a.get_services:
services = fw.config.get_services()
cmd.print_and_exit(" ".join(services))
elif a.new_service:
fw.config.new_service_dict(
a.new_service, FirewallClientServiceSettings().getSettingsDict()
)
elif a.new_service_from_file:
filename = os.path.basename(a.new_service_from_file)
dirname = os.path.dirname(a.new_service_from_file)
if dirname == "":
dirname = "./"
try:
obj = service_reader(filename, dirname)
except FirewallError as msg:
cmd.print_and_exit(
"Failed to load service file '%s': %s" % (a.new_service_from_file, msg),
msg.code,
)
except IOError as msg:
cmd.fail("Failed to load service file: %s" % msg)
if a.name:
obj.name = a.name
fw.config.new_service_dict(obj.name, obj.export_config_dict())
elif a.delete_service:
obj = fw.config.get_service(a.delete_service)
fw.config.remove_service(obj)
# remove service from all zones
zones = fw.config.get_zones()
for zone in zones:
_zone = fw.config.get_zone(zone)
_settings = FirewallClientZoneSettings(
fw.config.get_zone_config_dict(_zone)
)
if _settings.queryService(a.delete_service):
_settings.removeService(a.delete_service)
fw.config.set_zone_config_dict(_zone, _settings.getSettingsDict())
# remove service from all policies
for policy in fw.config.get_policy_objects():
_policy = fw.config.get_policy_object(policy)
_settings = FirewallClientPolicySettings(
fw.config.get_policy_object_config_dict(_policy)
)
if _settings.queryService(a.delete_service):
_settings.removeService(a.delete_service)
fw.config.set_policy_object_config_dict(
_policy, _settings.getSettingsDict()
)
elif a.load_service_defaults:
obj = fw.config.get_service(a.load_service_defaults)
fw.config.load_service_defaults(obj)
elif a.info_service:
service = fw.config.get_service(a.info_service)
settings = FirewallClientServiceSettings(
fw.config.get_service_config_dict(service)
)
cmd.print_service_info(a.info_service, settings)
sys.exit(0)
elif a.path_service:
obj = fw.config.get_service(a.path_service)
cmd.print_and_exit("%s/%s" % (obj.path, obj.filename))
# icmptypes
elif a.get_icmptypes:
icmptypes = fw.config.get_icmptypes()
cmd.print_and_exit(" ".join(icmptypes))
elif a.new_icmptype:
fw.config.new_icmptype(
a.new_icmptype, FirewallClientIcmpTypeSettings().settings
)
elif a.new_icmptype_from_file:
filename = os.path.basename(a.new_icmptype_from_file)
dirname = os.path.dirname(a.new_icmptype_from_file)
if dirname == "":
dirname = "./"
try:
obj = icmptype_reader(filename, dirname)
except FirewallError as msg:
cmd.print_and_exit(
"Failed to load icmptype file '%s': %s"
% (a.new_icmptype_from_file, msg),
msg.code,
)
except IOError as msg:
cmd.fail("Failed to load icmptype file: %s" % msg)
if a.name:
obj.name = a.name
fw.config.new_icmptype(obj.name, obj.export_config())
elif a.delete_icmptype:
obj = fw.config.get_icmptype(a.delete_icmptype)
fw.config.remove_icmptype(obj)
# remove icmpyte from all zones
zones = fw.config.get_zones()
for zone in zones:
_zone = fw.config.get_zone(zone)
_settings = FirewallClientZoneSettings(
fw.config.get_zone_config_dict(_zone)
)
if _settings.queryIcmpBlock(a.delete_icmptype):
_settings.removeIcmpBlock(a.delete_icmptype)
fw.config.set_zone_config_dict(_zone, _settings.getSettingsDict())
for policy in fw.config.get_policy_objects():
_policy = fw.config.get_policy_object(policy)
_settings = FirewallClientPolicySettings(
fw.config.get_policy_object_config_dict(_policy)
)
if _settings.queryIcmpBlock(a.delete_icmptype):
_settings.removeIcmpBlock(a.delete_icmptype)
fw.config.set_policy_object_config_dict(
_policy, _settings.getSettingsDict()
)
elif a.load_icmptype_defaults:
obj = fw.config.get_icmptype(a.load_icmptype_defaults)
fw.config.load_icmptype_defaults(obj)
elif a.info_icmptype:
icmptype = fw.config.get_icmptype(a.info_icmptype)
settings = FirewallClientIcmpTypeSettings(
list(fw.config.get_icmptype_config(icmptype))
)
cmd.print_icmptype_info(a.info_icmptype, settings)
sys.exit(0)
elif a.path_icmptype:
obj = fw.config.get_icmptype(a.path_icmptype)
cmd.print_and_exit("%s/%s" % (obj.path, obj.filename))
elif a.icmptype and options_icmptype:
icmptype = fw.config.get_icmptype(a.icmptype)
settings = FirewallClientIcmpTypeSettings(
list(fw.config.get_icmptype_config(icmptype))
)
if a.add_destination:
cmd.add_sequence(
a.add_destination,
settings.addDestination,
settings.queryDestination,
cmd.check_destination_ipv,
"'%s'",
)
fw.config.set_icmptype_config(icmptype, settings.settings)
elif a.remove_destination:
cmd.remove_sequence(
a.remove_destination,
settings.removeDestination,
settings.queryDestination,
cmd.check_destination_ipv,
"'%s'",
)
fw.config.set_icmptype_config(icmptype, settings.settings)
elif a.query_destination:
cmd.query_sequence(
a.query_destination,
settings.queryDestination,
cmd.check_destination_ipv,
"'%s'",
)
elif a.get_destinations:
l = settings.getDestinations()
if len(l) == 0:
l = ["ipv4", "ipv6"]
cmd.print_and_exit("\n".join(l))
elif a.set_description:
settings.setDescription(a.set_description)
fw.config.set_icmptype_config(icmptype, settings.settings)
elif a.get_description:
cmd.print_and_exit(settings.getDescription())
elif a.set_short:
settings.setShort(a.set_short)
fw.config.set_icmptype_config(icmptype, settings.settings)
elif a.get_short:
cmd.print_and_exit(settings.getShort())
else:
cmd.fail(parser.format_usage() + "Unknown option")
cmd.print_and_exit("success")
elif a.service and options_service:
service = fw.config.get_service(a.service)
settings = FirewallClientServiceSettings(
fw.config.get_service_config_dict(service)
)
if a.add_port:
cmd.add_sequence(
a.add_port,
settings.addPort,
settings.queryPort,
cmd.parse_port,
"%s/%s",
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.remove_port:
cmd.remove_sequence(
a.remove_port,
settings.removePort,
settings.queryPort,
cmd.parse_port,
"%s/%s",
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.query_port:
cmd.query_sequence(
a.query_port, settings.queryPort, cmd.parse_port, "%s/%s"
)
elif a.get_ports:
l = settings.getPorts()
cmd.print_and_exit(" ".join(["%s/%s" % (port[0], port[1]) for port in l]))
elif a.add_protocol:
cmd.add_sequence(
a.add_protocol,
settings.addProtocol,
settings.queryProtocol,
None,
"'%s'",
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.remove_protocol:
cmd.remove_sequence(
a.remove_protocol,
settings.removeProtocol,
settings.queryProtocol,
None,
"'%s'",
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.query_protocol:
cmd.query_sequence(a.query_protocol, settings.queryProtocol, None, "'%s'")
elif a.get_protocols:
l = settings.getProtocols()
cmd.print_and_exit(" ".join(["%s" % protocol for protocol in l]))
elif a.add_source_port:
cmd.add_sequence(
a.add_source_port,
settings.addSourcePort,
settings.querySourcePort,
cmd.parse_port,
"%s/%s",
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.remove_source_port:
cmd.remove_sequence(
a.remove_source_port,
settings.removeSourcePort,
settings.querySourcePort,
cmd.parse_port,
"%s/%s",
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.query_source_port:
cmd.query_sequence(
a.query_source_port, settings.querySourcePort, cmd.parse_port, "%s/%s"
)
elif a.get_source_ports:
l = settings.getSourcePorts()
cmd.print_and_exit(" ".join(["%s/%s" % (port[0], port[1]) for port in l]))
elif a.add_module:
cmd.add_sequence(
a.add_module, settings.addModule, settings.queryModule, None, "'%s'"
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.remove_module:
cmd.remove_sequence(
a.remove_module,
settings.removeModule,
settings.queryModule,
None,
"'%s'",
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.query_module:
cmd.query_sequence(a.query_module, settings.queryModule, None, "'%s'")
elif a.get_modules:
l = settings.getModules()
cmd.print_and_exit(" ".join(["%s" % module for module in l]))
elif a.set_destination:
cmd.add_sequence(
a.set_destination,
settings.setDestination,
settings.queryDestination,
cmd.parse_service_destination,
"%s:%s",
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.remove_destination:
# special case for removeDestination: Only ipv, no address
for ipv in a.remove_destination:
cmd.check_destination_ipv(ipv)
if ipv not in settings.getDestinations():
if len(a.remove_destination) > 1:
cmd.print_warning("Warning: NOT_ENABLED: '%s'" % ipv)
else:
code = FirewallError.get_code("NOT_ENABLED")
cmd.print_and_exit("Error: NOT_ENABLED: '%s'" % ipv, code)
else:
settings.removeDestination(ipv)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.query_destination:
cmd.query_sequence(
a.query_destination,
settings.queryDestination,
cmd.parse_service_destination,
"'%s'",
)
elif a.get_destinations:
l = settings.getDestinations()
cmd.print_and_exit(
" ".join(["%s:%s" % (dest[0], dest[1]) for dest in l.items()])
)
elif a.add_include:
cmd.add_sequence(
a.add_include, settings.addInclude, settings.queryInclude, None, "'%s'"
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.remove_include:
cmd.remove_sequence(
a.remove_include,
settings.removeInclude,
settings.queryInclude,
None,
"'%s'",
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.query_include:
cmd.query_sequence(a.query_include, settings.queryInclude, None, "'%s'")
elif a.get_includes:
l = settings.getIncludes()
cmd.print_and_exit(" ".join(["%s" % include for include in sorted(l)]))
elif a.add_helper:
cmd.add_sequence(
a.add_helper, settings.addHelper, settings.queryHelper, None, "'%s'"
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.remove_helper:
cmd.remove_sequence(
a.remove_helper,
settings.removeHelper,
settings.queryHelper,
None,
"'%s'",
)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.query_helper:
cmd.query_sequence(a.query_helper, settings.queryHelper, None, "'%s'")
elif a.get_service_helpers:
l = settings.getHelpers()
cmd.print_and_exit(" ".join(["%s" % helper for helper in sorted(l)]))
elif a.set_description:
settings.setDescription(a.set_description)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.get_description:
cmd.print_and_exit(settings.getDescription())
elif a.set_short:
settings.setShort(a.set_short)
fw.config.set_service_config_dict(service, settings.getSettingsDict())
elif a.get_short:
cmd.print_and_exit(settings.getShort())
else:
cmd.fail(parser.format_usage() + "Unknown option")
cmd.print_and_exit("success")
# ipsets
if a.get_ipsets:
ipsets = fw.config.get_ipsets()
cmd.print_and_exit(" ".join(sorted(ipsets)))
elif a.new_ipset:
if not a.type:
cmd.fail(parser.format_usage() + "No type specified.")
if a.type == "hash:mac" and a.family:
cmd.fail(
parser.format_usage()
+ "--family is not compatible with the hash:mac type"
)
settings = FirewallClientIPSetSettings()
settings.setType(a.type)
if a.option:
for opt in a.option:
settings.addOption(*cmd.parse_ipset_option(opt))
if a.family:
settings.addOption("family", a.family)
fw.config.new_ipset(a.new_ipset, settings.settings)
elif a.new_ipset_from_file:
filename = os.path.basename(a.new_ipset_from_file)
dirname = os.path.dirname(a.new_ipset_from_file)
if dirname == "":
dirname = "./"
try:
obj = ipset_reader(filename, dirname)
except FirewallError as msg:
cmd.print_and_exit(
"Failed to load ipset file '%s': %s" % (a.new_ipset_from_file, msg),
msg.code,
)
except IOError as msg:
cmd.fail("Failed to load ipset file: %s" % msg)
if a.name:
obj.name = a.name
fw.config.new_ipset(obj.name, obj.export_config())
elif a.delete_ipset:
ipset = fw.config.get_ipset(a.delete_ipset)
fw.config.remove_ipset(ipset)
elif a.load_ipset_defaults:
obj = fw.config.get_ipset(a.load_ipset_defaults)
fw.config.load_ipset_defaults(obj)
elif a.info_ipset:
ipset = fw.config.get_ipset(a.info_ipset)
settings = FirewallClientIPSetSettings(list(fw.config.get_ipset_config(ipset)))
cmd.print_ipset_info(a.info_ipset, settings)
sys.exit(0)
elif a.path_ipset:
obj = fw.config.get_ipset(a.path_ipset)
cmd.print_and_exit("%s/%s" % (obj.path, obj.filename))
elif a.ipset:
if a.add_entry:
ipset = fw.config.get_ipset(a.ipset)
settings = FirewallClientIPSetSettings(
list(fw.config.get_ipset_config(ipset))
)
cmd.add_sequence(
a.add_entry, settings.addEntry, settings.queryEntry, None, "'%s'"
)
fw.config.set_ipset_config(ipset, settings.settings)
elif a.remove_entry:
ipset = fw.config.get_ipset(a.ipset)
settings = FirewallClientIPSetSettings(
list(fw.config.get_ipset_config(ipset))
)
cmd.remove_sequence(
a.remove_entry, settings.removeEntry, settings.queryEntry, None, "'%s'"
)
fw.config.set_ipset_config(ipset, settings.settings)
elif a.query_entry:
ipset = fw.config.get_ipset(a.ipset)
settings = FirewallClientIPSetSettings(
list(fw.config.get_ipset_config(ipset))
)
cmd.query_sequence(a.query_entry, settings.queryEntry, None, "'%s'")
elif a.get_entries:
ipset = fw.config.get_ipset(a.ipset)
settings = FirewallClientIPSetSettings(
list(fw.config.get_ipset_config(ipset))
)
l = settings.getEntries()
cmd.print_and_exit("\n".join(l))
elif a.add_entries_from_file:
ipset = fw.config.get_ipset(a.ipset)
settings = FirewallClientIPSetSettings(
list(fw.config.get_ipset_config(ipset))
)
changed = False
for filename in a.add_entries_from_file:
try:
entries = cmd.get_ipset_entries_from_file(filename)
except IOError as msg:
message = "Failed to read file '%s': %s" % (filename, msg)
if len(a.add_entries_from_file) > 1:
cmd.print_warning(message)
else:
cmd.print_and_exit(message)
else:
old_entries = settings.getEntries()
entries_set = set()
for entry in old_entries:
entries_set.add(entry)
for entry in entries:
if entry not in entries_set:
old_entries.append(entry)
entries_set.add(entry)
changed = True
else:
cmd.print_if_verbose("Warning: ALREADY_ENABLED: %s" % entry)
if changed:
settings.setEntries(old_entries)
if changed:
fw.config.set_ipset_config(ipset, settings.settings)
elif a.remove_entries_from_file:
ipset = fw.config.get_ipset(a.ipset)
settings = FirewallClientIPSetSettings(
list(fw.config.get_ipset_config(ipset))
)
changed = False
for filename in a.remove_entries_from_file:
try:
entries = cmd.get_ipset_entries_from_file(filename)
except IOError as msg:
message = "Failed to read file '%s': %s" % (filename, msg)
if len(a.remove_entries_from_file) > 1:
cmd.print_warning(message)
else:
cmd.print_and_exit(message)
else:
old_entries = settings.getEntries()
entries_set = set()
for entry in old_entries:
entries_set.add(entry)
for entry in entries:
if entry in entries_set:
old_entries.remove(entry)
entries_set.discard(entry)
changed = True
else:
cmd.print_if_verbose("Warning: NOT_ENABLED: %s" % entry)
if changed:
settings.setEntries(old_entries)
if changed:
fw.config.set_ipset_config(ipset, settings.settings)
elif a.set_description:
ipset = fw.config.get_ipset(a.ipset)
settings = FirewallClientIPSetSettings(
list(fw.config.get_ipset_config(ipset))
)
settings.setDescription(a.set_description)
fw.config.set_ipset_config(ipset, settings.settings)
elif a.get_description:
ipset = fw.config.get_ipset(a.ipset)
settings = FirewallClientIPSetSettings(
list(fw.config.get_ipset_config(ipset))
)
cmd.print_and_exit(settings.getDescription())
elif a.set_short:
ipset = fw.config.get_ipset(a.ipset)
settings = FirewallClientIPSetSettings(
list(fw.config.get_ipset_config(ipset))
)
settings.setShort(a.set_short)
fw.config.set_ipset_config(ipset, settings.settings)
elif a.get_short:
ipset = fw.config.get_ipset(a.ipset)
settings = FirewallClientIPSetSettings(
list(fw.config.get_ipset_config(ipset))
)
cmd.print_and_exit(settings.getShort())
else:
cmd.fail(parser.format_usage() + "Unknown option")
cmd.print_and_exit("success")
# helper
elif a.get_helpers:
cmd.print_and_exit(" ".join(sorted(fw.config.get_helpers())))
elif a.new_helper:
if not a.module:
cmd.fail(parser.format_usage() + "No module specified.")
settings = FirewallClientHelperSettings()
settings.setModule(a.module)
if a.family:
settings.setFamily(a.family)
fw.config.new_helper(a.new_helper, settings.settings)
elif a.new_helper_from_file:
filename = os.path.basename(a.new_helper_from_file)
dirname = os.path.dirname(a.new_helper_from_file)
if dirname == "":
dirname = "./"
try:
obj = helper_reader(filename, dirname)
except FirewallError as msg:
cmd.print_and_exit(
"Failed to load helper file '%s': %s" % (a.new_helper_from_file, msg),
msg.code,
)
except IOError as msg:
cmd.fail("Failed to load helper file: %s" % msg)
if a.name:
obj.name = a.name
fw.config.new_helper(obj.name, obj.export_config())
elif a.delete_helper:
obj = fw.config.get_helper(a.delete_helper)
fw.config.remove_helper(obj)
elif a.load_helper_defaults:
obj = fw.config.get_helper(a.load_helper_defaults)
fw.config.load_helper_defaults(obj)
elif a.info_helper:
obj = fw.config.get_helper(a.info_helper)
settings = FirewallClientHelperSettings(list(fw.config.get_helper_config(obj)))
cmd.print_helper_info(a.info_helper, settings)
sys.exit(0)
elif a.path_helper:
obj = fw.config.get_helper(a.path_helper)
cmd.print_and_exit("%s/%s" % (obj.path, obj.filename))
elif a.helper:
obj = fw.config.get_helper(a.helper)
settings = FirewallClientHelperSettings(list(fw.config.get_helper_config(obj)))
if a.add_port:
cmd.add_sequence(
a.add_port,
settings.addPort,
settings.queryPort,
cmd.parse_port,
"%s/%s",
)
fw.config.set_helper_config(obj, settings.settings)
elif a.remove_port:
cmd.remove_sequence(
a.remove_port,
settings.removePort,
settings.queryPort,
cmd.parse_port,
"%s/%s",
)
fw.config.set_helper_config(obj, settings.settings)
elif a.query_port:
cmd.query_sequence(
a.query_port, settings.queryPort, cmd.parse_port, "%s/%s"
)
elif a.get_ports:
l = settings.getPorts()
cmd.print_and_exit(" ".join(["%s/%s" % (port[0], port[1]) for port in l]))
elif a.get_module:
cmd.print_and_exit(settings.getModule())
elif a.set_module:
settings.setModule(cmd.check_module(a.set_module))
fw.config.set_helper_config(obj, settings.settings)
elif a.get_family:
cmd.print_and_exit(settings.getFamily())
elif a.set_family:
settings.setFamily(cmd.check_helper_family(a.set_family[0]))
fw.config.set_helper_config(obj, settings.settings)
elif a.set_description:
settings.setDescription(a.set_description)
fw.config.set_helper_config(obj, settings.settings)
elif a.get_description:
cmd.print_and_exit(settings.getDescription())
elif a.set_short:
settings.setShort(a.set_short)
fw.config.set_helper_config(obj, settings.settings)
elif a.get_short:
cmd.print_and_exit(settings.getShort())
else:
cmd.fail(parser.format_usage() + "Unknown option")
# lockdown whitelist
elif options_lockdown_whitelist:
whitelist = fw.config.get_policies().lockdown_whitelist
# commands
if a.list_lockdown_whitelist_commands:
l = whitelist.get_commands()
cmd.print_and_exit("\n".join(l))
elif a.add_lockdown_whitelist_command:
cmd.add_sequence(
a.add_lockdown_whitelist_command,
whitelist.add_command,
whitelist.has_command,
None,
"'%s'",
)
elif a.remove_lockdown_whitelist_command:
cmd.remove_sequence(
a.remove_lockdown_whitelist_command,
whitelist.remove_command,
whitelist.has_command,
None,
"'%s'",
)
elif a.query_lockdown_whitelist_command:
cmd.query_sequence(
a.query_lockdown_whitelist_command, whitelist.has_command, None, "'%s'"
)
# contexts
elif a.list_lockdown_whitelist_contexts:
l = whitelist.get_contexts()
cmd.print_and_exit("\n".join(l))
elif a.add_lockdown_whitelist_context:
cmd.add_sequence(
a.add_lockdown_whitelist_context,
whitelist.add_context,
whitelist.has_context,
None,
"'%s'",
)
elif a.remove_lockdown_whitelist_context:
cmd.remove_sequence(
a.remove_lockdown_whitelist_context,
whitelist.remove_context,
whitelist.has_context,
None,
"'%s'",
)
elif a.query_lockdown_whitelist_context:
cmd.query_sequence(
a.query_lockdown_whitelist_context, whitelist.has_context, None, "'%s'"
)
# uids
elif a.list_lockdown_whitelist_uids:
l = whitelist.get_uids()
cmd.print_and_exit(" ".join(map(str, l)))
elif a.add_lockdown_whitelist_uid:
cmd.add_sequence(
a.add_lockdown_whitelist_uid,
whitelist.add_uid,
whitelist.has_uid,
None,
"'%s'",
)
elif a.remove_lockdown_whitelist_uid:
cmd.remove_sequence(
a.remove_lockdown_whitelist_uid,
whitelist.remove_uid,
whitelist.has_uid,
None,
"'%s'",
)
elif a.query_lockdown_whitelist_uid:
cmd.query_sequence(
a.query_lockdown_whitelist_uid, whitelist.has_uid, None, "'%s'"
)
# users
elif a.list_lockdown_whitelist_users:
l = whitelist.get_users()
cmd.print_and_exit("\n".join(l))
elif a.add_lockdown_whitelist_user:
cmd.add_sequence(
a.add_lockdown_whitelist_user,
whitelist.add_user,
whitelist.has_user,
None,
"'%s'",
)
elif a.remove_lockdown_whitelist_user:
cmd.remove_sequence(
a.remove_lockdown_whitelist_user,
whitelist.remove_user,
whitelist.has_user,
None,
"'%s'",
)
elif a.query_lockdown_whitelist_user:
cmd.query_sequence(
a.query_lockdown_whitelist_user, whitelist.has_user, None, "'%s'"
)
# apply whitelist changes
whitelist.write()
elif options_direct:
obj = fw.config.get_direct()
if a.add_passthrough:
if len(a.add_passthrough) < 2:
cmd.fail(
"usage: --direct --add-passthrough { ipv4 | ipv6 | eb } <args>"
)
cmd.print_msg(
obj.add_passthrough(
cmd.check_ipv(a.add_passthrough[0]), splitArgs(a.add_passthrough[1])
)
)
elif a.remove_passthrough:
if len(a.remove_passthrough) < 2:
cmd.fail(
"usage: --direct --remove-passthrough { ipv4 | ipv6 | eb } <args>"
)
obj.remove_passthrough(
cmd.check_ipv(a.remove_passthrough[0]),
splitArgs(a.remove_passthrough[1]),
)
elif a.query_passthrough:
if len(a.query_passthrough) < 2:
cmd.fail(
"usage: --direct --query-passthrough { ipv4 | ipv6 | eb } <args>"
)
cmd.print_query_result(
obj.query_passthrough(
cmd.check_ipv(a.query_passthrough[0]),
splitArgs(a.query_passthrough[1]),
)
)
sys.exit(0)
elif a.get_passthroughs:
rules = obj.get_passthroughs(cmd.check_ipv(a.get_passthroughs[0]))
for rule in rules:
cmd.print_msg(joinArgs(rule))
sys.exit(0)
elif a.get_all_passthroughs:
rules = obj.get_all_passthroughs()
for ipv in rules:
for rule in rules[ipv]:
cmd.print_msg("%s %s" % (ipv, joinArgs(rule)))
sys.exit(0)
elif a.add_chain:
obj.add_chain(cmd.check_ipv(a.add_chain[0]), a.add_chain[1], a.add_chain[2])
elif a.remove_chain:
obj.remove_chain(
cmd.check_ipv(a.remove_chain[0]), a.remove_chain[1], a.remove_chain[2]
)
elif a.query_chain:
cmd.print_query_result(
obj.query_chain(
cmd.check_ipv(a.query_chain[0]), a.query_chain[1], a.query_chain[2]
)
)
sys.exit(0)
elif a.get_chains:
cmd.print_and_exit(
" ".join(
obj.get_chains(cmd.check_ipv(a.get_chains[0]), a.get_chains[1])
)
)
sys.exit(0)
elif a.get_all_chains:
chains = obj.get_all_chains()
for ipv, table in chains:
for chain in chains[(ipv, table)]:
cmd.print_msg("%s %s %s" % (ipv, table, chain))
sys.exit(0)
elif a.add_rule:
if len(a.add_rule) < 5:
cmd.fail(
"usage: --direct --add-rule { ipv4 | ipv6 | eb } <table> <chain> <priority> <args>"
)
try:
priority = int(a.add_rule[3])
except ValueError:
cmd.fail(
"wrong priority\nusage: --direct --add-rule { ipv4 | ipv6 | eb } <table> <chain> <priority> <args>"
)
obj.add_rule(
cmd.check_ipv(a.add_rule[0]),
a.add_rule[1],
a.add_rule[2],
priority,
splitArgs(a.add_rule[4]),
)
elif a.remove_rule:
if len(a.remove_rule) < 5:
cmd.fail(
"usage: --direct --remove-rule { ipv4 | ipv6 | eb } <table> <chain> <priority> <args>"
)
try:
priority = int(a.remove_rule[3])
except ValueError:
cmd.fail(
"usage: --direct --remove-rule { ipv4 | ipv6 | eb } <table> <chain> <priority> <args>"
)
obj.remove_rule(
cmd.check_ipv(a.remove_rule[0]),
a.remove_rule[1],
a.remove_rule[2],
priority,
splitArgs(a.remove_rule[4]),
)
elif a.remove_rules:
if len(a.remove_rules) < 3:
cmd.fail(
"usage: --direct --remove-rules { ipv4 | ipv6 | eb } <table> <chain>"
)
obj.remove_rules(
cmd.check_ipv(a.remove_rules[0]), a.remove_rules[1], a.remove_rules[2]
)
elif a.query_rule:
if len(a.query_rule) < 5:
cmd.fail(
"usage: --direct --query-rule { ipv4 | ipv6 | eb } <table> <chain> <priority> <args>"
)
try:
priority = int(a.query_rule[3])
except ValueError:
cmd.fail(
"usage: --direct --query-rule { ipv4 | ipv6 | eb } <table> <chain> <priority> <args>"
)
cmd.print_query_result(
obj.query_rule(
cmd.check_ipv(a.query_rule[0]),
a.query_rule[1],
a.query_rule[2],
priority,
splitArgs(a.query_rule[4]),
)
)
sys.exit(0)
elif a.get_rules:
rules = obj.get_rules(
cmd.check_ipv(a.get_rules[0]), a.get_rules[1], a.get_rules[2]
)
for priority, rule in rules:
cmd.print_msg("%d %s" % (priority, joinArgs(rule)))
sys.exit(0)
elif a.get_all_rules:
rules = obj.get_all_rules()
for ipv, table, chain in rules:
for priority, rule in rules[(ipv, table, chain)]:
cmd.print_msg(
"%s %s %s %d %s" % (ipv, table, chain, priority, joinArgs(rule))
)
sys.exit(0)
obj.write()
# list everything
elif a.list_all_policies:
policies = fw.config.get_policy_objects()
for policy in policies:
fw_policy = fw.config.get_policy_object(policy)
fw_settings = FirewallClientPolicySettings(
fw.config.get_policy_object_config_dict(fw_policy)
)
cmd.print_policy_info(policy, fw_settings)
cmd.print_msg("")
sys.exit(0)
elif a.policy:
fw_policy = fw.config.get_policy_object(a.policy)
fw_settings = FirewallClientPolicySettings(
fw.config.get_policy_object_config_dict(fw_policy)
)
# ingress zones
if a.list_ingress_zones:
l = fw_settings.getIngressZones()
cmd.print_and_exit(" ".join(sorted(l)))
elif a.add_ingress_zone:
cmd.add_sequence(
a.add_ingress_zone,
fw_settings.addIngressZone,
fw_settings.queryIngressZone,
None,
"'%s'",
)
elif a.remove_ingress_zone:
cmd.remove_sequence(
a.remove_ingress_zone,
fw_settings.removeIngressZone,
fw_settings.queryIngressZone,
None,
"'%s'",
)
elif a.query_ingress_zone:
cmd.query_sequence(
a.query_ingress_zone, fw_settings.queryIngressZone, None, "'%s'"
)
# egress zones
if a.list_egress_zones:
l = fw_settings.getEgressZones()
cmd.print_and_exit(" ".join(sorted(l)))
elif a.add_egress_zone:
cmd.add_sequence(
a.add_egress_zone,
fw_settings.addEgressZone,
fw_settings.queryEgressZone,
None,
"'%s'",
)
elif a.remove_egress_zone:
cmd.remove_sequence(
a.remove_egress_zone,
fw_settings.removeEgressZone,
fw_settings.queryEgressZone,
None,
"'%s'",
)
elif a.query_egress_zone:
cmd.query_sequence(
a.query_egress_zone, fw_settings.queryEgressZone, None, "'%s'"
)
# priority
elif a.get_priority:
cmd.print_and_exit(str(fw_settings.getPriority()))
elif a.set_priority:
fw_settings.setPriority(a.set_priority)
# rich rules
if a.list_rich_rules:
l = fw_settings.getRichRules()
cmd.print_and_exit("\n".join(l))
elif a.add_rich_rule:
cmd.add_sequence(
a.add_rich_rule,
fw_settings.addRichRule,
fw_settings.queryRichRule,
None,
"'%s'",
)
elif a.remove_rich_rule:
cmd.remove_sequence(
a.remove_rich_rule,
fw_settings.removeRichRule,
fw_settings.queryRichRule,
None,
"'%s'",
)
elif a.query_rich_rule:
cmd.query_sequence(
a.query_rich_rule, fw_settings.queryRichRule, None, "'%s'"
)
# service
if a.list_services:
l = fw_settings.getServices()
cmd.print_and_exit(" ".join(sorted(l)))
elif a.add_service:
cmd.add_sequence(
a.add_service,
fw_settings.addService,
fw_settings.queryService,
None,
"'%s'",
)
elif a.remove_service_from_policy:
cmd.remove_sequence(
a.remove_service_from_policy,
fw_settings.removeService,
fw_settings.queryService,
None,
"'%s'",
)
elif a.query_service:
cmd.query_sequence(a.query_service, fw_settings.queryService, None, "'%s'")
# port
elif a.list_ports:
l = fw_settings.getPorts()
cmd.print_and_exit(
" ".join(
[
"%s/%s" % (port[0], port[1])
for port in sorted(
l, key=lambda x: (x[1], getPortRange(x[0])[0])
)
]
)
)
elif a.add_port:
cmd.add_sequence(
a.add_port,
fw_settings.addPort,
fw_settings.queryPort,
cmd.parse_port,
"%s/%s",
)
elif a.remove_port:
cmd.remove_sequence(
a.remove_port,
fw_settings.removePort,
fw_settings.queryPort,
cmd.parse_port,
"%s/%s",
)
elif a.query_port:
cmd.query_sequence(
a.query_port, fw_settings.queryPort, cmd.parse_port, "%s/%s"
)
# protocol
elif a.list_protocols:
l = fw_settings.getProtocols()
cmd.print_and_exit(" ".join(sorted(l)))
elif a.add_protocol:
cmd.add_sequence(
a.add_protocol,
fw_settings.addProtocol,
fw_settings.queryProtocol,
None,
"'%s'",
)
elif a.remove_protocol:
cmd.remove_sequence(
a.remove_protocol,
fw_settings.removeProtocol,
fw_settings.queryProtocol,
None,
"'%s'",
)
elif a.query_protocol:
cmd.query_sequence(
a.query_protocol, fw_settings.queryProtocol, None, "'%s'"
)
# source port
elif a.list_source_ports:
l = fw_settings.getSourcePorts()
cmd.print_and_exit(
" ".join(
[
"%s/%s" % (port[0], port[1])
for port in sorted(
l, key=lambda x: (x[1], getPortRange(x[0])[0])
)
]
)
)
elif a.add_source_port:
cmd.add_sequence(
a.add_source_port,
fw_settings.addSourcePort,
fw_settings.querySourcePort,
cmd.parse_port,
"%s/%s",
)
elif a.remove_source_port:
cmd.remove_sequence(
a.remove_source_port,
fw_settings.removeSourcePort,
fw_settings.querySourcePort,
cmd.parse_port,
"%s/%s",
)
elif a.query_source_port:
cmd.query_sequence(
a.query_source_port,
fw_settings.querySourcePort,
cmd.parse_port,
"%s/%s",
)
# masquerade
elif a.add_masquerade:
fw_settings.setMasquerade(True)
elif a.remove_masquerade:
fw_settings.setMasquerade(False)
elif a.query_masquerade:
cmd.print_query_result(fw_settings.getMasquerade())
# forward port
elif a.list_forward_ports:
l = fw_settings.getForwardPorts()
cmd.print_and_exit(
"\n".join(
[
"port=%s:proto=%s:toport=%s:toaddr=%s"
% (_port, _protocol, _toport, _toaddr)
for (_port, _protocol, _toport, _toaddr) in l
]
)
)
elif a.add_forward_port:
cmd.add_sequence(
a.add_forward_port,
fw_settings.addForwardPort,
fw_settings.queryForwardPort,
cmd.parse_forward_port,
"port=%s:proto=%s:toport=%s:toaddr=%s",
)
elif a.remove_forward_port:
cmd.remove_sequence(
a.remove_forward_port,
fw_settings.removeForwardPort,
fw_settings.queryForwardPort,
cmd.parse_forward_port,
"port=%s:proto=%s:toport=%s:toaddr=%s",
)
elif a.query_forward_port:
cmd.query_sequence(
a.query_forward_port,
fw_settings.queryForwardPort,
cmd.parse_forward_port,
"port=%s:proto=%s:toport=%s:toaddr=%s",
)
# block icmp
elif a.list_icmp_blocks:
l = fw_settings.getIcmpBlocks()
cmd.print_and_exit(" ".join(l))
elif a.add_icmp_block:
cmd.add_sequence(
a.add_icmp_block,
fw_settings.addIcmpBlock,
fw_settings.queryIcmpBlock,
None,
"'%s'",
)
elif a.remove_icmp_block:
cmd.remove_sequence(
a.remove_icmp_block,
fw_settings.removeIcmpBlock,
fw_settings.queryIcmpBlock,
None,
"'%s'",
)
elif a.query_icmp_block:
cmd.query_sequence(
a.query_icmp_block, fw_settings.queryIcmpBlock, None, "'%s'"
)
# policy target
elif a.get_target:
cmd.print_and_exit(fw_settings.getTarget())
elif a.set_target:
fw_settings.setTarget(a.set_target)
# list all policy settings
elif a.list_all:
cmd.print_policy_info(a.policy, fw_settings)
sys.exit(0)
elif a.set_description:
fw_settings.setDescription(a.set_description)
elif a.get_description:
cmd.print_and_exit(fw_settings.getDescription())
elif a.set_short:
fw_settings.setShort(a.set_short)
elif a.get_short:
cmd.print_and_exit(fw_settings.getShort())
fw.config.set_policy_object_config_dict(
fw_policy, fw_settings.getSettingsDict()
)
cmd.print_and_exit("success")
else:
if zone == "":
zone = fw.get_default_zone()
fw_zone = fw.config.get_zone(zone)
fw_settings = FirewallClientZoneSettings(
fw.config.get_zone_config_dict(fw_zone)
)
# interface
if a.list_interfaces:
l = fw_settings.getInterfaces()
cmd.print_and_exit(" ".join(l))
elif a.get_zone_of_interface:
for interface in a.get_zone_of_interface:
ret = []
for zone in fw.config.get_zones():
obj = fw.config.get_zone(zone)
if interface in obj.interfaces:
ret.append(obj.name)
if len(ret) > 1:
# Even it shouldn't happen, it's actually possible that
# the same interface is in several zone XML files
cmd.print_warning(
" ".join(ret)
+ " (ERROR: interface '%s' is in %s zone XML files, can be only in one)"
% (interface, len(ret))
)
if len(ret) == 1:
if len(a.get_zone_of_interface) > 1:
cmd.print_warning("%s: %s" % (interface, ret[0]))
else:
cmd.print_and_exit(ret[0])
else:
if len(a.get_zone_of_interface) > 1:
cmd.print_warning("%s: no zone" % interface)
else:
cmd.print_and_exit("no zone", 2)
elif a.change_interface:
for interface in a.change_interface:
for old_zone in fw.config.get_zones():
old_zone_obj = fw.config.get_zone(old_zone)
if interface in old_zone_obj.interfaces:
if old_zone_obj.name != zone:
old_zone_settings = FirewallClientZoneSettings(
fw.config.get_zone_config_dict(old_zone_obj)
)
old_zone_settings.removeInterface(
interface
) # remove from old
fw.config.set_zone_config_dict(
old_zone_obj, old_zone_settings.getSettingsDict()
)
fw_settings.addInterface(interface) # add to new
elif a.add_interface:
cmd.add_sequence(
a.add_interface,
fw_settings.addInterface,
fw_settings.queryInterface,
None,
"'%s'",
)
elif a.remove_interface:
cmd.remove_sequence(
a.remove_interface,
fw_settings.removeInterface,
fw_settings.queryInterface,
None,
"'%s'",
)
elif a.query_interface:
cmd.query_sequence(
a.query_interface, fw_settings.queryInterface, None, "'%s'"
)
# source
if a.list_sources:
sources = fw_settings.getSources()
cmd.print_and_exit(" ".join(sources))
elif a.get_zone_of_source:
for source in a.get_zone_of_source:
ret = []
for zone in fw.config.get_zones():
obj = fw.config.get_zone(zone)
if source in obj.sources:
ret.append(obj.name)
if len(ret) > 1:
# Even it shouldn't happen, it's actually possible that
# the same source is in several zone XML files
cmd.print_warning(
" ".join(ret)
+ " (ERROR: source '%s' is in %s zone XML files, can be only in one)"
% (source, len(ret))
)
if len(ret) == 1:
if len(a.get_zone_of_source) > 1:
cmd.print_warning("%s: %s" % (source, ret[0]))
else:
cmd.print_and_exit(ret[0])
else:
if len(a.get_zone_of_source) > 1:
cmd.print_warning("%s: no zone" % source)
else:
cmd.print_and_exit("no zone", 2)
elif a.change_source:
for source in a.change_source:
for old_zone in fw.config.get_zones():
old_zone_obj = fw.config.get_zone(old_zone)
if source in old_zone_obj.sources:
if old_zone_obj.name != zone:
old_zone_settings = FirewallClientZoneSettings(
fw.config.get_zone_config_dict(old_zone_obj)
)
old_zone_settings.removeSource(source) # remove from old
fw.config.set_zone_config_dict(
old_zone_obj, old_zone_settings.getSettingsDict()
)
fw_settings.addSource(source) # add to new
elif a.add_source:
cmd.add_sequence(
a.add_source,
fw_settings.addSource,
fw_settings.querySource,
None,
"'%s'",
)
elif a.remove_source:
cmd.remove_sequence(
a.remove_source,
fw_settings.removeSource,
fw_settings.querySource,
None,
"'%s'",
)
elif a.query_source:
cmd.query_sequence(a.query_source, fw_settings.querySource, None, "'%s'")
# rich rules
if a.list_rich_rules:
l = fw_settings.getRichRules()
cmd.print_and_exit("\n".join(l))
elif a.add_rich_rule:
cmd.add_sequence(
a.add_rich_rule,
fw_settings.addRichRule,
fw_settings.queryRichRule,
None,
"'%s'",
)
elif a.remove_rich_rule:
cmd.remove_sequence(
a.remove_rich_rule,
fw_settings.removeRichRule,
fw_settings.queryRichRule,
None,
"'%s'",
)
elif a.query_rich_rule:
cmd.query_sequence(
a.query_rich_rule, fw_settings.queryRichRule, None, "'%s'"
)
# service
if a.list_services:
l = fw_settings.getServices()
cmd.print_and_exit(" ".join(sorted(l)))
elif a.add_service:
cmd.add_sequence(
a.add_service,
fw_settings.addService,
fw_settings.queryService,
None,
"'%s'",
)
elif a.remove_service_from_zone:
cmd.remove_sequence(
a.remove_service_from_zone,
fw_settings.removeService,
fw_settings.queryService,
None,
"'%s'",
)
elif a.query_service:
cmd.query_sequence(a.query_service, fw_settings.queryService, None, "'%s'")
# port
elif a.list_ports:
l = fw_settings.getPorts()
cmd.print_and_exit(
" ".join(
[
"%s/%s" % (port[0], port[1])
for port in sorted(
l, key=lambda x: (x[1], getPortRange(x[0])[0])
)
]
)
)
elif a.add_port:
cmd.add_sequence(
a.add_port,
fw_settings.addPort,
fw_settings.queryPort,
cmd.parse_port,
"%s/%s",
)
elif a.remove_port:
cmd.remove_sequence(
a.remove_port,
fw_settings.removePort,
fw_settings.queryPort,
cmd.parse_port,
"%s/%s",
)
elif a.query_port:
cmd.query_sequence(
a.query_port, fw_settings.queryPort, cmd.parse_port, "%s/%s"
)
# protocol
elif a.list_protocols:
l = fw_settings.getProtocols()
cmd.print_and_exit(" ".join(sorted(l)))
elif a.add_protocol:
cmd.add_sequence(
a.add_protocol,
fw_settings.addProtocol,
fw_settings.queryProtocol,
None,
"'%s'",
)
elif a.remove_protocol:
cmd.remove_sequence(
a.remove_protocol,
fw_settings.removeProtocol,
fw_settings.queryProtocol,
None,
"'%s'",
)
elif a.query_protocol:
cmd.query_sequence(
a.query_protocol, fw_settings.queryProtocol, None, "'%s'"
)
# source port
elif a.list_source_ports:
l = fw_settings.getSourcePorts()
cmd.print_and_exit(
" ".join(
[
"%s/%s" % (port[0], port[1])
for port in sorted(
l, key=lambda x: (x[1], getPortRange(x[0])[0])
)
]
)
)
elif a.add_source_port:
cmd.add_sequence(
a.add_source_port,
fw_settings.addSourcePort,
fw_settings.querySourcePort,
cmd.parse_port,
"%s/%s",
)
elif a.remove_source_port:
cmd.remove_sequence(
a.remove_source_port,
fw_settings.removeSourcePort,
fw_settings.querySourcePort,
cmd.parse_port,
"%s/%s",
)
elif a.query_source_port:
cmd.query_sequence(
a.query_source_port,
fw_settings.querySourcePort,
cmd.parse_port,
"%s/%s",
)
# forward
elif a.add_forward:
fw_settings.setForward(True)
elif a.remove_forward:
fw_settings.setForward(False)
elif a.query_forward:
cmd.print_query_result(fw_settings.getForward())
# masquerade
elif a.add_masquerade:
fw_settings.setMasquerade(True)
elif a.remove_masquerade:
fw_settings.setMasquerade(False)
elif a.query_masquerade:
cmd.print_query_result(fw_settings.getMasquerade())
# forward port
elif a.list_forward_ports:
l = fw_settings.getForwardPorts()
cmd.print_and_exit(
"\n".join(
[
"port=%s:proto=%s:toport=%s:toaddr=%s"
% (_port, _protocol, _toport, _toaddr)
for (_port, _protocol, _toport, _toaddr) in l
]
)
)
elif a.add_forward_port:
cmd.add_sequence(
a.add_forward_port,
fw_settings.addForwardPort,
fw_settings.queryForwardPort,
cmd.parse_forward_port,
"port=%s:proto=%s:toport=%s:toaddr=%s",
)
elif a.remove_forward_port:
cmd.remove_sequence(
a.remove_forward_port,
fw_settings.removeForwardPort,
fw_settings.queryForwardPort,
cmd.parse_forward_port,
"port=%s:proto=%s:toport=%s:toaddr=%s",
)
elif a.query_forward_port:
cmd.query_sequence(
a.query_forward_port,
fw_settings.queryForwardPort,
cmd.parse_forward_port,
"port=%s:proto=%s:toport=%s:toaddr=%s",
)
# block icmp
elif a.list_icmp_blocks:
l = fw_settings.getIcmpBlocks()
cmd.print_and_exit(" ".join(l))
elif a.add_icmp_block:
cmd.add_sequence(
a.add_icmp_block,
fw_settings.addIcmpBlock,
fw_settings.queryIcmpBlock,
None,
"'%s'",
)
elif a.remove_icmp_block:
cmd.remove_sequence(
a.remove_icmp_block,
fw_settings.removeIcmpBlock,
fw_settings.queryIcmpBlock,
None,
"'%s'",
)
elif a.query_icmp_block:
cmd.query_sequence(
a.query_icmp_block, fw_settings.queryIcmpBlock, None, "'%s'"
)
# icmp block inversion
elif a.add_icmp_block_inversion:
fw_settings.addIcmpBlockInversion()
elif a.remove_icmp_block_inversion:
fw_settings.removeIcmpBlockInversion()
elif a.query_icmp_block_inversion:
cmd.print_query_result(fw_settings.queryIcmpBlockInversion())
# zone target
elif a.get_target:
cmd.print_and_exit(fw_settings.getTarget())
elif a.set_target:
fw_settings.setTarget(a.set_target)
# list all zone settings
elif a.list_all:
cmd.print_zone_info(
zone if zone else fw.get_default_zone(),
fw_settings,
default_zone=fw.get_default_zone(),
)
sys.exit(0)
# list everything
elif a.list_all_zones:
zones = fw.config.get_zones()
default_zone = fw.get_default_zone()
for zone in zones:
fw_zone = fw.config.get_zone(zone)
fw_settings = FirewallClientZoneSettings(
fw.config.get_zone_config_dict(fw_zone)
)
cmd.print_zone_info(zone, fw_settings, default_zone=default_zone)
cmd.print_msg("")
sys.exit(0)
elif a.set_description:
fw_settings.setDescription(a.set_description)
elif a.get_description:
cmd.print_and_exit(fw_settings.getDescription())
elif a.set_short:
fw_settings.setShort(a.set_short)
elif a.get_short:
cmd.print_and_exit(fw_settings.getShort())
# priority
elif a.get_priority:
cmd.print_and_exit(str(fw_settings.getPriority()))
elif a.set_priority:
fw_settings.setPriority(a.set_priority)
elif a.get_ingress_priority:
cmd.print_and_exit(str(fw_settings.getIngressPriority()))
elif a.set_ingress_priority:
fw_settings.setIngressPriority(a.set_ingress_priority)
elif a.get_egress_priority:
cmd.print_and_exit(str(fw_settings.getEgressPriority()))
elif a.set_egress_priority:
fw_settings.setEgressPriority(a.set_egress_priority)
fw.config.set_zone_config_dict(fw_zone, fw_settings.getSettingsDict())
cmd.print_and_exit("success")
except FirewallError as msg:
cmd.print_and_exit("%s" % msg, msg.code)
except Exception as msg:
cmd.fail("%s" % msg)
else:
cmd.print_and_exit("success")