Initial commit

.gitignore

@@ -0,0 +1,6 @@
+.vscode
+dist
+build
+.eggs
+stormshield.sns.sslclient.egg-info
+__pycache__

LICENCE

@@ -0,0 +1,13 @@
+Copyright 2018 Stormshield
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.

MANIFEST.in

@@ -0,0 +1,2 @@
+include stormshield/sns/bundle.ca
+include stormshield/sns/cmd.complete

README.md

@@ -0,0 +1,133 @@
+# pySNSAPI
+
+A Python client for the Stormshield Network Security appliance SSL API.
+
+## API usage
+
+```python
+from stormshield.sns.sslclient import SSLClient
+
+client = SSLClient(
+    host="10.0.0.254", port=443,
+    user='admin', password='password',
+    sslverifyhost=False)
+
+response = client.send_command("SYSTEM PROPERTY")
+
+if response:
+    model   = response.data['Result']['Model']
+    version = response.data['Result']['Version']
+
+    print("Model: {}".format(model))
+    print("Firmware version: {}".format(version))
+else:
+    print("Command failed: {}".format(response.output))
+
+client.disconnect()
+
+```
+
+### Command results
+
+Command results are available in text, xml or python structure formats:
+
+```python
+>>> response = client.send_command("CONFIG NTP SERVER LIST")
+
+>>> print(response.output)
+101 code=00a01000 msg="Begin" format="section_line"
+[Result]
+name=ntp1.stormshieldcs.eu keynum=none type=host
+name=ntp2.stormshieldcs.eu keynum=none type=host
+100 code=00a00100 msg="Ok"
+
+>>> print(response.xml)
+<?xml version="1.0"?>
+<nws code="100" msg="OK"><serverd ret="101" code="00a01000" msg="Begin"><data format="section_line"><section title="Result"><line><key name="name" value="ntp1.stormshieldcs.eu"/><key name="keynum" value="none"/><key name="type" value="host"/></line><line><key name="name" value="ntp2.stormshieldcs.eu"/><key name="keynum" value="none"/><key name="type" value="host"/></line></section></data></serverd><serverd ret="100" code="00a00100" msg="Ok"></serverd></nws>
+
+>>> print(response.data)
+{'Result': [{'name': 'ntp1.stormshieldcs.eu', 'keynum': 'none', 'type': 'host'}, {'name': 'ntp2.stormshieldcs.eu', 'keynum': 'none', 'type': 'host'}]}
+
+```
+
+The keys of the `data` property are case insensitive, `response.data['Result'][0]['name']` and `response.data['ReSuLt'][0]['NaMe']` will return the same value.
+
+Results token are also available via `response.parser.get()` method which accepts a default parameter to return if the token is not present.
+
+```python
+>>> print(response.output)
+101 code=00a01000 msg="Begin" format="section"
+[Server]
+1=dns1.google.com
+2=dns2.google.com
+100 code=00a00100 msg="Ok"
+
+>>> print(response.data['Server']['3'])
+Traceback (most recent call last):
+  File "<stdin>", line 1, in <module>
+  File "/usr/local/lib/python3.7/site-packages/requests/structures.py", line 52, in __getitem__
+    return self._store[key.lower()][1]
+KeyError: '3'
+
+>>> print(response.parser.get(section='Server', token='3', default=None))
+None
+
+```
+
+### File upload/download
+
+Files can be downloaded or uploaded by adding a redirection to a file with '>' or '<' at the end of the configuration command.
+
+```python
+>>> client.send_command("CONFIG BACKUP list=all > /tmp/mybackup.na")
+100 code=00a00100 msg="Ok"
+```
+
+## snscli
+
+ `snscli` is a python cli for executing configuration commands and scripts on Stormshield Network Security appliances.
+
+* Output format can be chosen between section/ini or xml
+* File upload and download available with adding `< upload` or `> download` at the end of the command
+* Client can execute script files using `--script` option.
+* Comments are allowed with `#`
+
+`$ snscli --host <utm>`
+
+`$ snscli --host <utm> --user admin --password admin --script config.script`
+
+Concerning the SSL validation:
+
+* For the first connection to a new appliance, ssl host name verification can be bypassed with `--no-sslverifyhost` option.
+* To connect to a known appliance with the default certificate use `--host <serial> --ip <ip address>` to validate the peer certificate.
+* If a custom CA and certificate is installed, use `--host myfirewall.tld --cabundle <ca.pem>`.
+* For client certificate authentication, the expected format is a pem file with the certificate and the unencrypted key concatenated.
+
+
+## Build
+
+`$ python3 setup.py sdist bdist_wheel`
+
+
+## Install
+
+`$ python3 setup.py install`
+
+
+## Tests
+
+Warning: tests require a remote SNS appliance.
+
+`$ APPLIANCE=10.0.0.254 python3 setup.py test`
+
+
+To run `snscli` from the source folder without install:
+
+`$ PYTHONPATH=. python3 ./bin/snscli --help`
+
+
+## Links
+
+* [Stormshield corporate website](https://www.stormshield.com)
+* [CLI commands reference guide](https://documentation.stormshield.eu/SNS/v3/en/Content/CLI_Serverd_Commands_reference_Guide_v3/Introduction.htm)
+

bin/snscli

@@ -0,0 +1,197 @@
+#!/usr/bin/python
+
+""" cli to connect to Stormshield Network Security appliances"""
+
+import sys
+import os
+import re
+import logging
+import readline
+import getpass
+import atexit
+import xml.dom.minidom
+import begin
+from pygments import highlight
+from pygments.lexers import XmlLexer
+from pygments.formatters import TerminalFormatter
+from colorlog import ColoredFormatter
+
+from stormshield.sns.sslclient import SSLClient, ServerError
+
+FORMATTER = ColoredFormatter(
+    "%(log_color)s%(levelname)-8s%(reset)s %(message)s",
+    datefmt=None,
+    reset=True,
+    log_colors={
+        'DEBUG': 'green',
+        'INFO': 'cyan',
+        'WARNING': 'yellow',
+        'ERROR': 'red',
+        'CRITICAL': 'red,bg_white'
+    },
+    secondary_log_colors={},
+    style='%'
+)
+
+EMPTY_RE = re.compile(r'^\s*$')
+
+def make_completer():
+    """ load completer for readline """
+    vocabulary = []
+    with open(SSLClient.get_completer(), "r") as completelist:
+        for line in completelist:
+            vocabulary.append(line.replace('.', ' ').strip('\n'))
+
+    def custom_complete(text, state):
+        results = [x for x in vocabulary if x.startswith(text)] + [None]
+        return results[state]
+    return custom_complete
+
+@begin.start(auto_convert=True, short_args=False, lexical_order=True)
+@begin.logging
+def main(host: 'Remote UTM' = None,
+         ip: 'Remote UTM ip' = None,
+         usercert: 'User certificate file' = None,
+         cabundle: 'CA bundle file' = None,
+         password: 'Password' = None,
+         port: 'Remote port' = 443,
+         user: 'User name' = 'admin',
+         sslverifypeer: 'Strict SSL CA check' = True,
+         sslverifyhost: 'Strict SSL host name check' = True,
+         credentials: 'Privilege list' = None,
+         script: 'Command script' = None,
+         outputformat: 'Output format (ini|xml)' = 'ini'):
+
+    for handler in logging.getLogger().handlers:
+        if handler.__class__ == logging.StreamHandler:
+            handler.setFormatter(FORMATTER)
+
+    if script is not None:
+        try:
+            script = open(script, 'r')
+        except Exception as exception:
+            logging.error("Can't open script file - %s", str(exception))
+            sys.exit(1)
+
+    if outputformat not in ['ini', 'xml']:
+        logging.error("Unknown output format")
+        sys.exit(1)
+
+    if host is None:
+        logging.error("No host provided")
+        sys.exit(1)
+
+    if password is None and usercert is None:
+        password = getpass.getpass()
+
+    try:
+        client = SSLClient(
+            host=host, ip=ip, port=port, user=user, password=password,
+            sslverifypeer=sslverifypeer, sslverifyhost=sslverifyhost,
+            credentials=credentials,
+            usercert=usercert, cabundle=cabundle, autoconnect=False)
+    except Exception as exception:
+        logging.error(str(exception))
+        sys.exit(1)
+
+    try:
+        client.connect()
+    except Exception as exception:
+        search = re.search(r'doesn\'t match \'(.*)\'', str(exception))
+        if search:
+            logging.error(("Appliance name can't be verified, to force connection "
+                           "use \"--host %s --ip %s\" or \"--no-sslverifyhost\" "
+                           "options"), search.group(1), host)
+        else:
+            logging.error(str(exception))
+        sys.exit(1)
+
+    # disconnect gracefuly at exit
+    atexit.register(client.disconnect)
+
+    if script is not None:
+        for cmd in script.readlines():
+            cmd = cmd.strip('\r\n')
+            print(cmd)
+            if cmd.startswith('#'):
+                continue
+            if EMPTY_RE.match(cmd):
+                continue
+            try:
+                response = client.send_command(cmd)
+            except Exception as exception:
+                logging.error(str(exception))
+                sys.exit(1)
+            if outputformat == 'xml':
+                print(highlight(xml.dom.minidom.parseString(response.xml).toprettyxml(),
+                                XmlLexer(), TerminalFormatter()))
+            else:
+                print(response.output)
+        sys.exit(0)
+
+    # Start cli
+
+    # load history
+    histfile = os.path.join(os.path.expanduser("~"), ".sslclient_history")
+    try:
+        readline.read_history_file(histfile)
+        readline.set_history_length(1000)
+    except FileNotFoundError:
+        pass
+
+    def save_history(histfile):
+        try:
+            readline.write_history_file(histfile)
+        except:
+            logging.warning("Can't write history")
+
+    atexit.register(save_history, histfile)
+
+    # load auto-complete
+    readline.parse_and_bind('tab: complete')
+    readline.set_completer_delims('')
+    readline.set_completer(make_completer())
+
+    while True:
+        try:
+            cmd = input("> ")
+        except EOFError:
+            break
+
+        # skip comments
+        if cmd.startswith('#'):
+            continue
+
+        try:
+            response = client.send_command(cmd)
+        except ServerError as exception:
+            # do not log error on QUIT
+            if "quit".startswith(cmd.lower()) \
+               and str(exception) == "Server disconnected":
+                sys.exit(0)
+            logging.error(str(exception))
+            sys.exit(1)
+        except Exception as exception:
+            logging.error(str(exception))
+            sys.exit(1)
+
+        if response.ret == client.SRV_RET_DOWNLOAD:
+            filename = input("File to save: ")
+            try:
+                client.download(filename)
+                logging.info("File downloaded")
+            except Exception as exception:
+                logging.error(str(exception))
+        elif response.ret == client.SRV_RET_UPLOAD:
+            filename = input("File to upload: ")
+            try:
+                client.upload(filename)
+                logging.info("File uploaded")
+            except Exception as exception:
+                logging.error(str(exception))
+        else:
+            if outputformat == 'xml':
+                print(highlight(xml.dom.minidom.parseString(response.xml).toprettyxml(),
+                                XmlLexer(), TerminalFormatter()))
+            else:
+                print(response.output)

examples/example.script

@@ -0,0 +1,3 @@
+#system information
+SYSTEM PROPERTY
+

examples/getproperty.py

@@ -0,0 +1,39 @@
+#!/usr/bin/python
+
+"""
+This example show how to connect to a SNS appliance, send a command
+to get appliance properties and parse the result to extract the
+appliance model and firmware version.
+"""
+
+import getpass
+
+from stormshield.sns.sslclient import SSLClient
+
+# user input
+host = input("Appliance ip address: ")
+user = input("User:")
+password = getpass.getpass("Password: ")
+
+# connect to the appliance
+client = SSLClient(
+    host=host, port=443,
+    user=user, password=password,
+    sslverifyhost=False)
+
+# request appliance properties
+response = client.send_command("SYSTEM PROPERTY")
+
+if response:
+    #get value using parser get method
+    model = response.parser.get(section='Result', token='Model')
+    # get value with direct access to data
+    version = response.data['Result']['Version']
+
+    print("")
+    print("Model: {}".format(model))
+    print("Firmware version: {}".format(version))
+else:
+    print("Command failed: {}".format(response.output))
+
+client.disconnect()