Skip to content

Commit

Permalink
make linters hapy
Browse files Browse the repository at this point in the history
Signed-off-by: Alejandro Hernández Cordero <[email protected]>
  • Loading branch information
ahcorde committed Jan 20, 2025
1 parent 6cdfbe1 commit 3c1d7b9
Show file tree
Hide file tree
Showing 5 changed files with 111 additions and 52 deletions.
4 changes: 4 additions & 0 deletions zenoh_security_configuration/package.xml
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,10 @@
<depend>python3-jsonschema</depend>
<depend>python3-yaml</depend>

<test_depend>ament_flake8</test_depend>
<test_depend>ament_xmllint</test_depend>
<test_depend>python3-pytest</test_depend>

<export>
<build_type>ament_python</build_type>
</export>
Expand Down
3 changes: 2 additions & 1 deletion zenoh_security_configuration/setup.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import os
from glob import glob
import os

from setuptools import setup

package_name = 'zenoh_security_configuration'
Expand Down
23 changes: 23 additions & 0 deletions zenoh_security_configuration/test/test_flake8.py
Original file line number Diff line number Diff line change
@@ -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'
23 changes: 23 additions & 0 deletions zenoh_security_configuration/test/test_xmllint.py
Original file line number Diff line number Diff line change
@@ -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'
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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:
Expand All @@ -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'.
Expand All @@ -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:
Expand All @@ -82,127 +84,132 @@ 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:])

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:
Expand All @@ -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)

0 comments on commit 3c1d7b9

Please sign in to comment.