syslog: T4251: Add TLS support to syslog

Add TLS support for remote syslog by extending the CLI and backend to support configuration of CA certificates, client certificates, keys, and authentication modes.

This update integrates with the PKI subsystem for certificate management, ensures proper validation of protocol settings when TLS is enabled, and generates secure rsyslog configuration for forwarding logs over TLS.

1 files changed, 93 insertions, 2 deletions

diff --git a/src/conf_mode/system_syslog.py b/src/conf_mode/system_syslog.py
index c1a8baa1d..82be09e4f 100755
--- a/src/conf_mode/system_syslog.py
+++ b/src/conf_mode/system_syslog.py
@@ -15,15 +15,22 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
+import shutil
 
 from sys import exit
 
 from vyos.base import Warning
 from vyos.config import Config
 from vyos.configverify import verify_vrf
+from vyos.configverify import verify_pki_certificate
+from vyos.configverify import verify_pki_ca_certificate
 from vyos.defaults import systemd_services
 from vyos.utils.network import is_addr_assigned
 from vyos.utils.process import call
+from vyos.utils.dict import dict_search
+from vyos.utils.file import write_file
+from vyos.pki import wrap_certificate
+from vyos.pki import wrap_private_key
 from vyos.template import render
 from vyos.template import is_ipv4
 from vyos.template import is_ipv6
@@ -31,12 +38,75 @@ from vyos import ConfigError
 from vyos import airbag
 airbag.enable()
 
+cert_dir = '/etc/rsyslog.d/certs'
 rsyslog_conf = '/run/rsyslog/rsyslog.conf'
 logrotate_conf = '/etc/logrotate.d/vyos-rsyslog'
 
 systemd_socket = 'syslog.socket'
 systemd_service = systemd_services['syslog']
 
+
+def _cleanup_tls_certs():
+    if os.path.exists(cert_dir):
+        shutil.rmtree(cert_dir, ignore_errors=True)
+
+
+def _remote_has_tls(remote_options):
+    return 'tls' in remote_options and 'enable' in remote_options['tls']
+
+
+def _verify_tls_remote_options(remote, remote_options, syslog):
+    auth_mode = dict_search('tls.auth_mode', remote_options)
+    certificate = dict_search('tls.certificate', remote_options)
+    ca_certificate = dict_search('tls.ca_certificate', remote_options)
+
+    if auth_mode != "anon" and not ca_certificate:
+        raise ConfigError(
+            f'Option "ca-certificate" is required for remote "{remote}" when TLS is enabled with auth-mode "{auth_mode}"!'
+        )
+
+    if certificate:
+        verify_pki_certificate(syslog, certificate, no_password_protected=True)
+
+    if ca_certificate:
+        verify_pki_ca_certificate(syslog, ca_certificate)
+
+    permitted_peers = dict_search('tls.permitted_peers', remote_options)
+    if not permitted_peers:
+        if auth_mode == "fingerprint":
+            raise ConfigError(
+                f'Auth mode "fingerprint" for remote "{remote}" requires "permitted-peers" to be configured!'
+            )
+        elif auth_mode == "name":
+            raise ConfigError(
+                f'Auth mode "name" for remote "{remote}" requires "permitted-peers" to specify allowed subject names!'
+            )
+
+
+def _save_tls_certificates_for_remote(syslog, remote_options):
+    ca_certificate = remote_options['tls'].get('ca_certificate')
+    ca_cert_file_path = None
+    if ca_certificate:
+        ca_cert_file_path = os.path.join(cert_dir, f'{ca_certificate}.pem')
+        pki_ca = syslog['pki']['ca'][ca_certificate]
+
+        ca_cert = wrap_certificate(pki_ca['certificate'])
+        write_file(ca_cert_file_path, ca_cert)
+    remote_options['tls']['ca_certificate_path'] = ca_cert_file_path
+
+    cert_name = remote_options['tls'].get('certificate')
+    cert_file_path = cert_key_path = None
+    if cert_name:
+        cert_file_path = os.path.join(cert_dir, f'{cert_name}.pem')
+        cert_key_path = os.path.join(cert_dir, f'{cert_name}.key')
+        pki_cert = syslog['pki']['certificate'][cert_name]
+
+        write_file(cert_file_path, wrap_certificate(pki_cert['certificate']))
+        write_file(cert_key_path, wrap_private_key(pki_cert['private']['key']))
+
+    remote_options['tls']['certificate_path'] = cert_file_path
+    remote_options['tls']['certificate_key_path'] = cert_key_path
+
 def get_config(config=None):
     if config:
         conf = config
@@ -46,8 +116,13 @@ def get_config(config=None):
     if not conf.exists(base):
         return None
 
-    syslog = conf.get_config_dict(base, key_mangling=('-', '_'),
-                                  get_first_key=True, no_tag_node_value_mangle=True)
+    syslog = conf.get_config_dict(
+        base,
+        key_mangling=('-', '_'),
+        get_first_key=True,
+        no_tag_node_value_mangle=True,
+        with_pki=True,
+    )
 
     syslog.update({ 'logrotate' : logrotate_conf })
 
@@ -97,7 +172,18 @@ def verify(syslog):
                     raise ConfigError(f'Source-address "{source_address}" does not match '\
                                       f'address-family of remote "{remote}"!')
 
+            if _remote_has_tls(remote_options):
+                _verify_tls_remote_options(remote, remote_options, syslog)
+
+                if 'protocol' in remote_options and remote_options['protocol'] == 'udp':
+                    Warning(
+                        f'TLS is enabled for remote "{remote}", but protocol is set to UDP. TLS is only supported with protocol TCP!'
+                    )
+
+
 def generate(syslog):
+    _cleanup_tls_certs()
+
     if not syslog:
         if os.path.exists(rsyslog_conf):
             os.unlink(rsyslog_conf)
@@ -106,6 +192,11 @@ def generate(syslog):
 
         return None
 
+    if 'remote' in syslog:
+        for _, remote_options in syslog['remote'].items():
+            if _remote_has_tls(remote_options):
+                _save_tls_certificates_for_remote(syslog, remote_options)
+
     render(rsyslog_conf, 'rsyslog/rsyslog.conf.j2', syslog)
     render(logrotate_conf, 'rsyslog/logrotate.j2', syslog)
     return None