diff --git a/zenoh_security_configuration/package.xml b/zenoh_security_configuration/package.xml index 3f6c03c5..6a2893b9 100644 --- a/zenoh_security_configuration/package.xml +++ b/zenoh_security_configuration/package.xml @@ -10,6 +10,10 @@ python3-jsonschema python3-yaml + ament_flake8 + ament_xmllint + python3-pytest + ament_python diff --git a/zenoh_security_configuration/setup.py b/zenoh_security_configuration/setup.py index 35c02461..58566a16 100644 --- a/zenoh_security_configuration/setup.py +++ b/zenoh_security_configuration/setup.py @@ -1,5 +1,6 @@ -import os from glob import glob +import os + from setuptools import setup package_name = 'zenoh_security_configuration' diff --git a/zenoh_security_configuration/test/test_flake8.py b/zenoh_security_configuration/test/test_flake8.py new file mode 100644 index 00000000..b9042824 --- /dev/null +++ b/zenoh_security_configuration/test/test_flake8.py @@ -0,0 +1,23 @@ +# Copyright 2019 Open Source Robotics Foundation, Inc. +# +# 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. + +from ament_flake8.main import main +import pytest + + +@pytest.mark.flake8 +@pytest.mark.linter +def test_flake8(): + rc = main(argv=[]) + assert rc == 0, 'Found errors' diff --git a/zenoh_security_configuration/test/test_xmllint.py b/zenoh_security_configuration/test/test_xmllint.py new file mode 100644 index 00000000..ca32830a --- /dev/null +++ b/zenoh_security_configuration/test/test_xmllint.py @@ -0,0 +1,23 @@ +# Copyright 2019 Open Source Robotics Foundation, Inc. +# +# 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. + +from ament_xmllint.main import main +import pytest + + +@pytest.mark.linter +@pytest.mark.xmllint +def test_xmllint(): + rc = main(argv=['.']) + assert rc == 0, 'Found errors' diff --git a/zenoh_security_configuration/zenoh_security_configuration/zenoh_security_configuration.py b/zenoh_security_configuration/zenoh_security_configuration/zenoh_security_configuration.py index 4d0d22aa..dd3ef467 100644 --- a/zenoh_security_configuration/zenoh_security_configuration/zenoh_security_configuration.py +++ b/zenoh_security_configuration/zenoh_security_configuration/zenoh_security_configuration.py @@ -13,21 +13,22 @@ # limitations under the License. import argparse -import em from io import StringIO import os import pathlib import sys + from ament_index_python.packages import get_package_share_directory +import em + + class ZenohSecutiryConfigGenerator: """ZenohSecutiryConfigGenerator generates Zenoh secutiry configurations.""" def __init__(self): - """ - Construct ZenohConfigGenerator. - """ - self.zenoh_cfg_file_extension = "json5" + """Construct ZenohConfigGenerator.""" + self.zenoh_cfg_file_extension = 'json5' def generate_router_config(self, data, zenoh_type): try: @@ -41,9 +42,9 @@ def generate_router_config(self, data, zenoh_type): ) template_path = pathlib.Path(os.path.join( - get_package_share_directory("zenoh_security_configuration"), - "templates", - zenoh_type + ".json5" + get_package_share_directory('zenoh_security_configuration'), + 'templates', + zenoh_type + '.json5' )) with template_path.open('r') as h: @@ -57,7 +58,8 @@ def generate_router_config(self, data, zenoh_type): interpreter.shutdown() interpreter = None - def generate_zenoh_config(self, output_dir, router_config, zenoh_type, encoding: str = 'utf-8'): + def generate_zenoh_config(self, output_dir, router_config, + zenoh_type, encoding: str = 'utf-8'): """ Generate Zenoh bridge configs and output to directory 'output_dir'. @@ -68,11 +70,11 @@ def generate_zenoh_config(self, output_dir, router_config, zenoh_type, encoding: """ write_filepath = os.path.join( - output_dir, zenoh_type + "." + output_dir, zenoh_type + '.' + self.zenoh_cfg_file_extension, ) output_file = pathlib.Path(write_filepath) - print(f"Generated Zenoh secutiry configuration at {write_filepath}") + print(f'Generated Zenoh security configuration at {write_filepath}') if output_file.exists(): existing_content = output_file.read_text(encoding=encoding) if existing_content == router_config: @@ -82,117 +84,120 @@ def generate_zenoh_config(self, output_dir, router_config, zenoh_type, encoding: output_file.write_text(router_config, encoding=encoding) + def main(argv=sys.argv): """Entrypoint.""" parser = argparse.ArgumentParser( - description="Generate Zenoh security configurations", + description='Generate Zenoh security configurations', formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) - subparsers = parser.add_subparsers(help='help for subcommand', dest="subcommand") + subparsers = parser.add_subparsers(help='Help for subcommand', dest='subcommand') parser_path = subparsers.add_parser('paths', help='Use path') parser_enclave = subparsers.add_parser('enclave', help='Use enclave directory') parser.add_argument( - "-o", - "--output", + '-o', + '--output', required=True, type=str, - help="Output directory for Zenoh bridge configurations", + help='Output directory for Zenoh bridge configurations', ) parser.add_argument( - "-l", - "--listen_endpoint", + '-l', + '--listen_endpoint', required=False, type=str, - default ="tls/localhost:7447", - help="The list of endpoints to listen on (See https://docs.rs/zenoh/latest/zenoh/config/struct.EndPoint.html)", + default='tls/localhost:7447', + help='The list of endpoints to listen on. ' + + '(See https://docs.rs/zenoh/latest/zenoh/config/struct.EndPoint.html)', ) parser.add_argument( - "-c", - "--connect_endpoint", + '-c', + '--connect_endpoint', required=False, type=str, - default ="tls/localhost:7447", - help="The list of endpoints to connect to. (See https://docs.rs/zenoh/latest/zenoh/config/struct.EndPoint.html)", + default='tls/localhost:7447', + help='The list of endpoints to connect to. ' + + '(See https://docs.rs/zenoh/latest/zenoh/config/struct.EndPoint.html)', ) parser.add_argument( - "-p", - "--protocols", + '-p', + '--protocols', nargs='*', required=False, - choices=["tcp", "tls"], + choices=['tcp', 'tls'], default=['tls'], - help="Protocols chooices", + help='Protocols chooices', ) parser.add_argument( - "-t", - "--type", + '-t', + '--type', type=str, required=True, - choices=["router", "peer"], + choices=['router', 'peer'], default=['router'], - help="Set router or peer", + help='Set router or peer', ) parser_path.add_argument( - "--root_ca_certificate", + '--root_ca_certificate', type=str, required=True, default='', - help="Path to the certificate of the certificate authority used to validate " \ - "either the server or the client's keys and certificates" + help='Path to the certificate of the certificate authority used' + + "to validate either the server or the client's keys and certificates" ) parser_path.add_argument( - "--listen_private_key", + '--listen_private_key', type=str, required=True, default='', - help="Path to the TLS listening side private key" + help='Path to the TLS listening side private key' ) parser_path.add_argument( - "--listen_certificate", + '--listen_certificate', type=str, required=True, default='', - help="Path to the TLS listening side public certificate" + help='Path to the TLS listening side public certificate' ) parser_path.add_argument( - "--connect_private_key", + '--connect_private_key', type=str, required=True, default='', - help="Path to the TLS connecting side private key" + help='Path to the TLS connecting side private key' ) parser_path.add_argument( - "--connect_certificate", + '--connect_certificate', type=str, required=True, default='', - help="Path to the TLS connecting side certificate" + help='Path to the TLS connecting side certificate' ) parser_enclave.add_argument( - "--enclave_path", + '--enclave_path', type=str, required=True, default='', - help="Enclave path" + help='Enclave path' ) parser_enclave.add_argument( - "--enclave_name", + '--enclave_name', type=str, required=True, default='', - help="Enclave name" + help='Enclave name' ) args = parser.parse_args(argv[1:]) @@ -200,9 +205,11 @@ def main(argv=sys.argv): if args.enclave_name is not None: if args.enclave_name[0] == '/': args.enclave_name = args.enclave_name[1:] - root_ca_certificate = os.path.join(args.enclave_path, "public", "ca.cert.pem") - listen_private_key = os.path.join(args.enclave_path, "enclaves", args.enclave_name, "key.pem") - listen_certificate = os.path.join(args.enclave_path, "enclaves", args.enclave_name, "cert.pem") + root_ca_certificate = os.path.join(args.enclave_path, 'public', 'ca.cert.pem') + listen_private_key = os.path.join(args.enclave_path, 'enclaves', + args.enclave_name, 'key.pem') + listen_certificate = os.path.join(args.enclave_path, 'enclaves', + args.enclave_name, 'cert.pem') connect_private_key = listen_private_key connect_certificate = listen_certificate else: @@ -226,5 +233,6 @@ def main(argv=sys.argv): router_config = zenoh_sec_gen.generate_router_config(data, args.type) zenoh_sec_gen.generate_zenoh_config(args.output, router_config, args.type) -if __name__ == "__main__": + +if __name__ == '__main__': main(sys.argv)