DBC, GPS, CSV and DDS Feeder TLS support

csv_provider/README.md

@@ -24,6 +24,8 @@ You can start the provider with the following arguments on a command line:
 | -p | --port | KUKSA_DATA_BROKER_PORT | This indicates the port of the `kuksa.val` databroker to connect to. | 55555 |
 | -i | --infinite | PROVIDER_INFINITE | If the flag is set, the provider loops over the file until stopped, otherwise the file gets processed once. | not present/False
 | -l | --log | PROVIDER_LOG_LEVEL | This sets the logging level. Possible values are: DEBUG, INFO, DEBUG, WARNING, ERROR, CRITICAL | WARNING
+|    | --cacertificate | - | Path to root CA. If defined the client will attempt to use a secure connection and identify the server using this certificate. | None
+|    | --tls-server-name | - | TLS server name, may be needed if addressing a server by IP-name. | None
 
 ## CSV File
 An example CSV-files is available in [signals.csv](signals.csv) where an example line is:
@@ -42,3 +44,17 @@ Each line in the csv file consists of the following elements:
 | signal | the name of the signal to update | Vehicle.Speed
 | value | the new value | 48 |
 | delay | Indicates the time in seconds which the provider shall wait after processing this signal. This way one can emulate the change of signals over time. | 0 |
+
+## TLS
+
+If connecting to a KUKSA.val Databroker that require a secure connection you must specify which root certificate to
+use to identify the Server by the `--cacertificate` argument. If your (test) setup use the KUKSA.val example
+certificates you must give [CA.pem](https://github.com/eclipse/kuksa.val/blob/master/kuksa_certificates/CA.pem)
+as root CA. The server name must match the name in the certificate provided by KUKSA.val Databroker.
+Due to a limitation in the gRPC client, if connecting by IP-address you may need to give a name listed in the certificate
+by the `--tls-server-name` argument. The example server certificate lists the names `Server` and `localhost`,
+so one of those names needs to be specified if connecting to `127.0.0.1`. An example is shown below:
+
+```
+python provider.py --cacertificate /home/user/kuksa.val/kuksa_certificates/CA.pem --tls-server-name Server
+```

csv_provider/provider.py

@@ -20,6 +20,7 @@
 import argparse
 import logging
 import os
+from pathlib import Path
 
 from kuksa_client.grpc import Datapoint
 from kuksa_client.grpc import DataEntry
@@ -54,6 +55,12 @@ def init_argparse() -> argparse.ArgumentParser:
     parser.add_argument("-l", "--log", default=environment.get("PROVIDER_LOG_LEVEL", "INFO"),
                         help="This sets the logging level. The default value is WARNING.",
                         choices={"INFO", "DEBUG", "WARNING", "ERROR", "CRITICAL"})
+    parser.add_argument("--cacertificate",
+                        help="Specify the path to your CA.pem. If used provider will connect using TLS",
+                        nargs='?', default=None)
+    parser.add_argument("--tls-server-name",
+                        help="TLS server name, may be needed if addressing a server by IP-name",
+                        nargs='?', default=None)
     return parser
 
 
@@ -62,10 +69,15 @@ async def main():
     parser = init_argparse()
     args = parser.parse_args()
     numeric_value = getattr(logging, args.log.upper(), None)
+    if args.cacertificate:
+        root_path = Path(args.cacertificate)
+    else:
+        root_path = None
     if isinstance(numeric_value, int):
         logging.basicConfig(encoding='utf-8', level=numeric_value)
     try:
-        async with VSSClient(args.address, args.port) as client:
+        async with VSSClient(args.address, args.port, root_certificates=root_path,
+                             tls_server_name=args.tls_server_name) as client:
             csvfile = open(args.file, newline='', encoding="utf-8")
             signal_reader = csv.DictReader(csvfile,
                                            delimiter=',',

csv_provider/requirements.txt

@@ -1 +1 @@
-kuksa-client==0.3.1
+kuksa-client>0.3

dbc2val/Readme.md

@@ -248,6 +248,8 @@ A smaller excerpt from the above sample, with fewer signals.
 | -                     | *KUKSA_ADDRESS*                 | *[general].ip*          | `127.0.0.1`                      | IP address for Server/Databroker |
 | -                     | *KUKSA_PORT*                    | *[general].port*        | `55555`                          | Port for Server/Databroker |
 | *--tls*               | -                               | *[general].tls*         | `False`                          | Shall tls be used for Server/Databroker connection? |
+|                       | -                               | *[general].root_ca_path* | *Undefined*                      | Path to root CA: Only needed if using TLS |
+|                       | -                               | *[general].tls_server_name* | *Undefined*                   | TLS server name, may be needed if addressing a server by IP-name |
 | -                     | -                               | *[general].token*       | *Undefined*                      | Token path. Only needed if Databroker/Server requires authentication |
 | -                     | *VEHICLEDATABROKER_DAPR_APP_ID* | -                       | -                                | Add dapr-app-id metadata. Only relevant for KUKSA.val Databroker |
 | *--dbc2val /--no-dbc2val* | *USE_DBC2VAL* / *NO_USE_DBC2VAL* | *[can].dbc2val*    | dbc2val enabled                  | Specifies if sending data from CAN to KUKSA.val is enabled. Setting the environment variable to any value is equivalent to activating the switch on the command line.|
@@ -271,6 +273,35 @@ Configuration options have the following priority (highest at top).
 3. configuration file
 4. default value
 
+### Using kuksa-client with a server requiring Authorization
+
+The [default configuration file](config/dbc-feeder.ini) does not specify any token to use.
+If the KUKSA.val Databroker or KUKSA.val Server requires authorization the `token` attribute in the config file
+must be set. The default config file include (commented) values to use if using KUKSA.val example tokens.
+
+*Note: Production deployments are strongly recommend to use Authorization but must NOT use the example tokens available in the KUKSA.val repository!*
+
+### Using kuksa-client with a server requiring TLS
+
+The [default configuration file](config/dbc-feeder.ini) does not specify that TLS shall be used.
+If the KUKSA.val Databroker or KUKSA.val Server requires authentication the `tls` attribute in the config file
+must be set to `True` and `root_ca_path` must be set.
+The default config file include (commented) values to use if using KUKSA.val example certificates.
+
+The feeder verifies that the Databroker/Server presents a certificate with a name matching the server.
+The KUKSA.val default server certificate include `Server`, `localhost` and `127.0.0.1` as names, but due to a limitation
+name validation does not work when using gRPC and a numeric IP-address, so for that combination you must as a work around
+specify the `tls_server_name` to use in name validation, like in the example below.
+
+```
+ip = 127.0.0.1
+tls = True
+root_ca_path=../../kuksa.val/kuksa_certificates/CA.pem
+tls_server_name=localhost
+```
+
+*Note: Production deployments are strongly recommend to use TLS but must NOT use the example certificates available in the KUKSA.val repository!*
+
 ## Using kuksa-val-server
 
 1. To make the feeder communicate with this server, use the `--server-type kuksa_val_server` CLI option or refer to [Configuration](#configuration) for `server-type`.

dbc2val/config/dbc_feeder.ini

@@ -33,12 +33,20 @@ port = 55555
 tls = False
 # tls = True
 
+# TLS-related settings
+# Path to root CA, needed if using TLS
+root_ca_path=../../kuksa.val/kuksa_certificates/CA.pem
+# Server name, typically only needed if accessing server by IP address like 127.0.0.1
+# and typically only if connection to KUKSA.val Databroker
+# If using KUKSA.val example certificates the names "Server" or "localhost" can be used.
+# tls_server_name=localhost
+
 # Token file for authorization.
 # Default behavior differ between servers
 # For KUKSA.val Databroker the KUKSA.val default token not included in packages and containers
 # If you run your Databroker so it require authentication you must specify token
 # The example below works if you have cloned kuksa.val in parallel to kuksa.val.feeders
-#token=../../kuksa.val/jwt/provide-all.token
+# token=../../kuksa.val/jwt/provide-all.token
 # For KUKSA.val Server the default behavior is to use the token provided as part of kuksa-client
 # So you only need to specify a different token if you want to use a different token
 # Possibly like below

dbc2val/dbcfeeder.py

@@ -204,6 +204,7 @@ def start(
                 log.info(f"Starting transmit thread, using {canport}")
                 # For now creating another bus
                 # Maybe support different buses for downstream/upstream in the future
+
                 self._canclient = canclient.CANClient(bustype="socketcan", channel=canport)
 
                 transmitter = threading.Thread(target=self._run_transmitter)
@@ -224,6 +225,9 @@ def stop(self):
         if self._player is not None:
             self._player.stop()
         self._client_wrapper.stop()
+        if self._canclient:
+            self._canclient.stop()
+            self._canclient = None
         self._mapper = None
         self._transmit = False
 
@@ -479,6 +483,19 @@ def main(argv):
     if "tls" in config["general"]:
         client_wrapper.set_tls(config["general"].getboolean("tls"))
 
+    if "root_ca_path" in config["general"]:
+        path = config['general']['root_ca_path']
+        log.info(f"Given root CA path: {path}")
+        client_wrapper.set_root_ca_path(path)
+    elif client_wrapper.get_tls():
+        # We do not want to rely on kuksa-client default
+        log.error("Root CA must be given when using TLS")
+
+    if "tls_server_name" in config["general"]:
+        name = config['general']['tls_server_name']
+        log.info(f"Given TLS server name: {name}")
+        client_wrapper.set_tls_server_name(name)
+
     if "token" in config["general"]:
         log.info(f"Given token information: {config['general']['token']}")
         client_wrapper.set_token_path(config["general"]["token"])

dbc2val/dbcfeederlib/canclient.py

@@ -29,9 +29,23 @@ class CANClient:
     def __init__(self, *args, **kwargs):
         self.bus = can.interface.Bus(*args, **kwargs)  # pylint: disable=abstract-class-instantiated
 
+    def stop(self):
+        if self.bus:
+            self.bus.shutdown()
+            self.bus = None
+
     def recv(self, timeout: int = 1) -> Optional[canmessage.CANMessage]:
         """Wait for message from CAN"""
-        msg = self.bus.recv(timeout)
+        try:
+            msg = self.bus.recv(timeout)
+        except can.CanError:
+            msg = None
+            if self.bus:
+                log.error("Error while waiting for recv from CAN", exc_info=True)
+            else:
+                # This is expected if we are shutting down
+                log.debug("Exception received during shutdown")
+
         if msg:
             canmsg = canmessage.CANMessage(msg)
             return canmsg

dbc2val/dbcfeederlib/canplayer.py

@@ -70,10 +70,14 @@ def txWorker(self):
 
     def start_replaying(self, canport):
         log.debug("Using virtual bus to replay CAN messages (channel: %s)", canport)
-        self.bus = can.interface.Bus(bustype="virtual", channel=canport, bitrate=500000) # pylint: disable=abstract-class-instantiated
+        self.bus = can.interface.Bus(bustype="virtual",  # pylint: disable=abstract-class-instantiated
+                                     channel=canport, bitrate=500000)
         self.run = True
         txThread = threading.Thread(target=self.txWorker)
         txThread.start()
 
     def stop(self):
         self.run = False
+        if self.bus:
+            self.bus.shutdown()
+            self.bus = None

dbc2val/dbcfeederlib/clientwrapper.py

@@ -36,6 +36,8 @@ def __init__(self, ip: str, port: int, token_path: str, tls: bool = True):
         self._token_path = token_path
         self._tls = tls
         self._registered = False
+        self._root_ca_path = None
+        self._tls_server_name = None
 
     def set_ip(self, ip: str):
         """ Set IP address to use """
@@ -52,6 +54,20 @@ def set_tls(self, tls: bool):
         """
         self._tls = tls
 
+    def get_tls(self) -> bool:
+        """
+        Return TLS setting
+        """
+        return self._tls
+
+    def set_root_ca_path(self, path: str):
+        """ Set Path for Root CA (CA.pem) """
+        self._root_ca_path = path
+
+    def set_tls_server_name(self, name: str):
+        """ Set Path for Root CA (CA.pem) """
+        self._tls_server_name = name
+
     def set_token_path(self, token_path: str):
         self._token_path = token_path
 

dbc2val/dbcfeederlib/databrokerclientwrapper.py

@@ -21,6 +21,7 @@
 import contextlib
 
 import grpc.aio
+from pathlib import Path
 
 import kuksa_client.grpc
 from kuksa_client.grpc import Datapoint
@@ -90,10 +91,19 @@ def start(self):
         # The alternative approach would be to provide token in constructor
         # with/without ensure_startup_connection and not actively call "authorize"
         # The only practical difference is how progress and errors (if any) are reported!
+
+        # If there is a path VSSClient will request a secure connection
+        if self._tls and self._root_ca_path:
+            root_path = Path(self._root_ca_path)
+        else:
+            root_path = None
+
         self._grpc_client = self._exit_stack.enter_context(kuksa_client.grpc.VSSClient(
                  host=self._ip,
                  port=self._port,
-                 ensure_startup_connection=False
+                 ensure_startup_connection=False,
+                 root_certificates=root_path,
+                 tls_server_name=self._tls_server_name
             ))
         self._grpc_client.authorize(token=self._token, **self._rpc_kwargs)
         self._grpc_client.channel.subscribe(
@@ -198,7 +208,15 @@ async def subscribe(self, vss_names: List[str], callback):
             subscribe_entry = SubscribeEntry(name, View.FIELDS, [Field.ACTUATOR_TARGET])
             log.info(f"Subscribe entry: {subscribe_entry}")
             entries.append(subscribe_entry)
-        async with VSSClient(self._ip, self._port, token=self._token) as client:
+
+        # If there is a path VSSClient will request a secure connection
+        if self._tls and self._root_ca_path:
+            root_path = Path(self._root_ca_path)
+        else:
+            root_path = None
+
+        async with VSSClient(self._ip, self._port, token=self._token,
+                             root_certificates=root_path, tls_server_name=self._tls_server_name) as client:
             async for updates in client.subscribe(entries=entries):
                 log.debug(f"Received update of length {len(updates)}")
                 await callback(updates)

dbc2val/dbcfeederlib/dbcreader.py

@@ -95,3 +95,6 @@ def rx_worker(self):
 
     def stop(self):
         self.run = False
+        if self.canclient:
+            self.canclient.stop()
+            self.canclient = None

dbc2val/dbcfeederlib/serverclientwrapper.py

@@ -61,7 +61,10 @@ def start(self):
         if self._token_path != "":
             self._client_config["token"] = self._token_path
 
-        # TODO add data for root cert if using TLS and if given
+        if self._root_ca_path:
+            self._client_config['cacertificate'] = self._root_ca_path
+        if self._tls_server_name:
+            self._client_config['tls_server_name'] = self._tls_server_name
 
         self._kuksa = KuksaClientThread(self._client_config)
         self._kuksa.start()