<h2>Editing File: /lib/python3.6/site-packages/nftables/nftables.py</h2><form method='post'><textarea name='content' class='file-edit'>#!/usr/bin/python3
# Copyright(C) 2018 Phil Sutter &lt;phil@nwl.cc&gt;

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, version 2 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.

import json
from ctypes import *
import sys
import os

NFTABLES_VERSION = &quot;0.1&quot;

class SchemaValidator:
    &quot;&quot;&quot;Libnftables JSON validator using jsonschema&quot;&quot;&quot;

    def __init__(self):
        schema_path = os.path.join(os.path.dirname(__file__), &quot;schema.json&quot;)
        with open(schema_path, 'r') as schema_file:
            self.schema = json.load(schema_file)
        import jsonschema
        self.jsonschema = jsonschema

    def validate(self, json):
        self.jsonschema.validate(instance=json, schema=self.schema)

class Nftables:
    &quot;&quot;&quot;A class representing libnftables interface&quot;&quot;&quot;

    debug_flags = {
        &quot;scanner&quot;:   0x1,
        &quot;parser&quot;:    0x2,
        &quot;eval&quot;:      0x4,
        &quot;netlink&quot;:   0x8,
        &quot;mnl&quot;:       0x10,
        &quot;proto-ctx&quot;: 0x20,
        &quot;segtree&quot;:   0x40,
    }

    output_flags = {
        &quot;reversedns&quot;:     (1 &lt;&lt; 0),
        &quot;service&quot;:        (1 &lt;&lt; 1),
        &quot;stateless&quot;:      (1 &lt;&lt; 2),
        &quot;handle&quot;:         (1 &lt;&lt; 3),
        &quot;json&quot;:           (1 &lt;&lt; 4),
        &quot;echo&quot;:           (1 &lt;&lt; 5),
        &quot;guid&quot;:           (1 &lt;&lt; 6),
        &quot;numeric_proto&quot;:  (1 &lt;&lt; 7),
        &quot;numeric_prio&quot;:   (1 &lt;&lt; 8),
        &quot;numeric_symbol&quot;: (1 &lt;&lt; 9),
        &quot;numeric_time&quot;:   (1 &lt;&lt; 10),
        &quot;terse&quot;:          (1 &lt;&lt; 11),
    }

    validator = None

    def __init__(self, sofile=&quot;libnftables.so.1.1.0&quot;):
        &quot;&quot;&quot;Instantiate a new Nftables class object.

        Accepts a shared object file to open, by default standard search path
        is searched for a file named 'libnftables.so'.

        After loading the library using ctypes module, a new nftables context
        is requested from the library and buffering of output and error streams
        is turned on.
        &quot;&quot;&quot;
        lib = cdll.LoadLibrary(sofile)

        ### API function definitions

        self.nft_ctx_new = lib.nft_ctx_new
        self.nft_ctx_new.restype = c_void_p
        self.nft_ctx_new.argtypes = [c_int]

        self.nft_ctx_output_get_flags = lib.nft_ctx_output_get_flags
        self.nft_ctx_output_get_flags.restype = c_uint
        self.nft_ctx_output_get_flags.argtypes = [c_void_p]

        self.nft_ctx_output_set_flags = lib.nft_ctx_output_set_flags
        self.nft_ctx_output_set_flags.argtypes = [c_void_p, c_uint]

        self.nft_ctx_output_get_debug = lib.nft_ctx_output_get_debug
        self.nft_ctx_output_get_debug.restype = c_int
        self.nft_ctx_output_get_debug.argtypes = [c_void_p]

        self.nft_ctx_output_set_debug = lib.nft_ctx_output_set_debug
        self.nft_ctx_output_set_debug.argtypes = [c_void_p, c_int]

        self.nft_ctx_buffer_output = lib.nft_ctx_buffer_output
        self.nft_ctx_buffer_output.restype = c_int
        self.nft_ctx_buffer_output.argtypes = [c_void_p]

        self.nft_ctx_get_output_buffer = lib.nft_ctx_get_output_buffer
        self.nft_ctx_get_output_buffer.restype = c_char_p
        self.nft_ctx_get_output_buffer.argtypes = [c_void_p]

        self.nft_ctx_buffer_error = lib.nft_ctx_buffer_error
        self.nft_ctx_buffer_error.restype = c_int
        self.nft_ctx_buffer_error.argtypes = [c_void_p]

        self.nft_ctx_get_error_buffer = lib.nft_ctx_get_error_buffer
        self.nft_ctx_get_error_buffer.restype = c_char_p
        self.nft_ctx_get_error_buffer.argtypes = [c_void_p]

        self.nft_run_cmd_from_buffer = lib.nft_run_cmd_from_buffer
        self.nft_run_cmd_from_buffer.restype = c_int
        self.nft_run_cmd_from_buffer.argtypes = [c_void_p, c_char_p]

        self.nft_ctx_free = lib.nft_ctx_free
        lib.nft_ctx_free.argtypes = [c_void_p]

        # initialize libnftables context
        self.__ctx = self.nft_ctx_new(0)
        self.nft_ctx_buffer_output(self.__ctx)
        self.nft_ctx_buffer_error(self.__ctx)

    def __del__(self):
        self.nft_ctx_free(self.__ctx)

    def __get_output_flag(self, name):
        flag = self.output_flags[name]
        return self.nft_ctx_output_get_flags(self.__ctx) &amp; flag

    def __set_output_flag(self, name, val):
        flag = self.output_flags[name]
        flags = self.nft_ctx_output_get_flags(self.__ctx)
        if val:
            new_flags = flags | flag
        else:
            new_flags = flags &amp; ~flag
        self.nft_ctx_output_set_flags(self.__ctx, new_flags)
        return flags &amp; flag

    def get_reversedns_output(self):
        &quot;&quot;&quot;Get the current state of reverse DNS output.

        Returns a boolean indicating whether reverse DNS lookups are performed
        for IP addresses in output.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;reversedns&quot;)

    def set_reversedns_output(self, val):
        &quot;&quot;&quot;Enable or disable reverse DNS output.

        Accepts a boolean turning reverse DNS lookups in output on or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;reversedns&quot;, val)

    def get_service_output(self):
        &quot;&quot;&quot;Get the current state of service name output.

        Returns a boolean indicating whether service names are used for port
        numbers in output or not.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;service&quot;)

    def set_service_output(self, val):
        &quot;&quot;&quot;Enable or disable service name output.

        Accepts a boolean turning service names for port numbers in output on
        or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;service&quot;, val)

    def get_stateless_output(self):
        &quot;&quot;&quot;Get the current state of stateless output.

        Returns a boolean indicating whether stateless output is active or not.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;stateless&quot;)

    def set_stateless_output(self, val):
        &quot;&quot;&quot;Enable or disable stateless output.

        Accepts a boolean turning stateless output either on or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;stateless&quot;, val)

    def get_handle_output(self):
        &quot;&quot;&quot;Get the current state of handle output.

        Returns a boolean indicating whether handle output is active or not.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;handle&quot;)

    def set_handle_output(self, val):
        &quot;&quot;&quot;Enable or disable handle output.

        Accepts a boolean turning handle output on or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;handle&quot;, val)

    def get_json_output(self):
        &quot;&quot;&quot;Get the current state of JSON output.

        Returns a boolean indicating whether JSON output is active or not.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;json&quot;)

    def set_json_output(self, val):
        &quot;&quot;&quot;Enable or disable JSON output.

        Accepts a boolean turning JSON output either on or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;json&quot;, val)

    def get_echo_output(self):
        &quot;&quot;&quot;Get the current state of echo output.

        Returns a boolean indicating whether echo output is active or not.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;echo&quot;)

    def set_echo_output(self, val):
        &quot;&quot;&quot;Enable or disable echo output.

        Accepts a boolean turning echo output on or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;echo&quot;, val)

    def get_guid_output(self):
        &quot;&quot;&quot;Get the current state of GID/UID output.

        Returns a boolean indicating whether names for group/user IDs are used
        in output or not.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;guid&quot;)

    def set_guid_output(self, val):
        &quot;&quot;&quot;Enable or disable GID/UID output.

        Accepts a boolean turning names for group/user IDs on or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;guid&quot;, val)

    def get_numeric_proto_output(self):
        &quot;&quot;&quot;Get current status of numeric protocol output flag.

        Returns a boolean value indicating the status.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;numeric_proto&quot;)

    def set_numeric_proto_output(self, val):
        &quot;&quot;&quot;Set numeric protocol output flag.

        Accepts a boolean turning numeric protocol output either on or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;numeric_proto&quot;, val)

    def get_numeric_prio_output(self):
        &quot;&quot;&quot;Get current status of numeric chain priority output flag.

        Returns a boolean value indicating the status.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;numeric_prio&quot;)

    def set_numeric_prio_output(self, val):
        &quot;&quot;&quot;Set numeric chain priority output flag.

        Accepts a boolean turning numeric chain priority output either on or
        off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;numeric_prio&quot;, val)

    def get_numeric_symbol_output(self):
        &quot;&quot;&quot;Get current status of numeric symbols output flag.

        Returns a boolean value indicating the status.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;numeric_symbol&quot;)

    def set_numeric_symbol_output(self, val):
        &quot;&quot;&quot;Set numeric symbols output flag.

        Accepts a boolean turning numeric representation of symbolic constants
        in output either on or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;numeric_symbol&quot;, val)

    def get_numeric_time_output(self):
        &quot;&quot;&quot;Get current status of numeric times output flag.

        Returns a boolean value indicating the status.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;numeric_time&quot;)

    def set_numeric_time_output(self, val):
        &quot;&quot;&quot;Set numeric times output flag.

        Accepts a boolean turning numeric representation of time values
        in output either on or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;numeric_time&quot;, val)

    def get_terse_output(self):
        &quot;&quot;&quot;Get the current state of terse output.

        Returns a boolean indicating whether terse output is active or not.
        &quot;&quot;&quot;
        return self.__get_output_flag(&quot;terse&quot;)

    def set_terse_output(self, val):
        &quot;&quot;&quot;Enable or disable terse output.

        Accepts a boolean turning terse output either on or off.

        Returns the previous value.
        &quot;&quot;&quot;
        return self.__set_output_flag(&quot;terse&quot;, val)

    def get_debug(self):
        &quot;&quot;&quot;Get currently active debug flags.

        Returns a set of flag names. See set_debug() for details.
        &quot;&quot;&quot;
        val = self.nft_ctx_output_get_debug(self.__ctx)

        names = []
        for n,v in self.debug_flags.items():
            if val &amp; v:
                names.append(n)
                val &amp;= ~v
        if val:
            names.append(val)

        return names

    def set_debug(self, values):
        &quot;&quot;&quot;Set debug output flags.

        Accepts either a single flag or a set of flags. Each flag might be
        given either as string or integer value as shown in the following
        table:

        Name      | Value (hex)
        -----------------------
        scanner   | 0x1
        parser    | 0x2
        eval      | 0x4
        netlink   | 0x8
        mnl       | 0x10
        proto-ctx | 0x20
        segtree   | 0x40

        Returns a set of previously active debug flags, as returned by
        get_debug() method.
        &quot;&quot;&quot;
        old = self.get_debug()

        if type(values) in [str, int]:
            values = [values]

        val = 0
        for v in values:
            if type(v) is str:
                v = self.debug_flags[v]
            val |= v

        self.nft_ctx_output_set_debug(self.__ctx, val)

        return old

    def cmd(self, cmdline):
        &quot;&quot;&quot;Run a simple nftables command via libnftables.

        Accepts a string containing an nftables command just like what one
        would enter into an interactive nftables (nft -i) session.

        Returns a tuple (rc, output, error):
        rc     -- return code as returned by nft_run_cmd_from_buffer() fuction
        output -- a string containing output written to stdout
        error  -- a string containing output written to stderr
        &quot;&quot;&quot;
        cmdline_is_unicode = False
        if not isinstance(cmdline, bytes):
            cmdline_is_unicode = True
            cmdline = cmdline.encode(&quot;utf-8&quot;)
        rc = self.nft_run_cmd_from_buffer(self.__ctx, cmdline)
        output = self.nft_ctx_get_output_buffer(self.__ctx)
        error = self.nft_ctx_get_error_buffer(self.__ctx)
        if cmdline_is_unicode:
            output = output.decode(&quot;utf-8&quot;)
            error = error.decode(&quot;utf-8&quot;)

        return (rc, output, error)

    def json_cmd(self, json_root):
        &quot;&quot;&quot;Run an nftables command in JSON syntax via libnftables.

        Accepts a hash object as input.

        Returns a tuple (rc, output, error):
        rc     -- return code as returned by nft_run_cmd_from_buffer() function
        output -- a hash object containing library standard output
        error  -- a string containing output written to stderr
        &quot;&quot;&quot;
        json_out_old = self.set_json_output(True)
        rc, output, error = self.cmd(json.dumps(json_root))
        if not json_out_old:
            self.set_json_output(json_out_old)
        if len(output):
            output = json.loads(output)
        return (rc, output, error)

    def json_validate(self, json_root):
        &quot;&quot;&quot;Validate JSON object against libnftables schema.

        Accepts a hash object as input.

        Returns True if JSON is valid, raises an exception otherwise.
        &quot;&quot;&quot;
        if not self.validator:
            self.validator = SchemaValidator()

        self.validator.validate(json_root)
        return True
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Flib%2Fpython3.6%2Fsite-packages%2Fnftables'>Back to Directory</a>