Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

tests: adding external logger tests #6

Merged
merged 7 commits into from
Dec 15, 2023
Merged

tests: adding external logger tests #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
9ca19ab
move to utils file
epsilon-0 Dec 14, 2023
29e57db
refactor common code
epsilon-0 Dec 14, 2023
d6efaf7
add syslogng tests
epsilon-0 Dec 15, 2023
7749ed4
add mutual TLS tests
epsilon-0 Dec 15, 2023
88b7684
mutual tls example
epsilon-0 Dec 15, 2023
666a36e
move start/stop_server to setUp/tearDown
epsilon-0 Dec 15, 2023
fdb908e
use create_default_context for mutual auth
epsilon-0 Dec 15, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions README.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,17 +36,23 @@ Usage
secure={cafile='/path/to/ca/file'})
logger.addHandler(handler2)

# with custom SSLContext
context = ssl.create_default_context(cafile='/path/to/ca/file')
# with custom SSLContext (e.g. for mutual TLS authentication)
context = ssl.create_default_context(
purpose=ssl.Purpose.SERVER_AUTH, cafile="/path/to/ca/file"
)
context.load_cert_chain(
certfile="/path/to/client/cert.pem",
keyfile="/path/to/client/priv.key",
)
handler3 = TLSSysLogHandler(address=('secure-logging.example.com', 6514),
socktype=socket.SOCK_STREAM,
secure=context)
logger.addHandler(handler3)

# or allow TLS without verification
# or allow TLS without verification (not recommended)
handler4 = TLSSysLogHandler(address=('secure-logging.example.com', 6514),
socktype=socket.SOCK_STREAM,
secure="noverify")
logger.addHandler(handler4)

logger.info('Hello World!')
logger.info('Hello, World!')
135 changes: 4 additions & 131 deletions tests/test_regress_basic.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,48 +1,20 @@
import datetime
import ipaddress
import multiprocessing
import concurrent.futures
import os
import tempfile
from time import sleep
from unittest import TestCase, mock
from unittest import mock
import uuid

from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import hashes, serialization

import ssl
import logging
import socket
import inspect

from tlssysloghandler import TLSSysLogHandler

SOCKET_PORT = int(os.environ.get("SOCKET_PORT", 56712))
from test_util import SOCKET_PORT, TestCertManager

SOCKET_TIMEOUT = 5
SOCKET_BUFFERSIZE = 1024

RSA_PUBLIC_EXPONENT = 65537
RSA_KEY_SIZE = 2048

# logger = logging.getLogger(__name__)


class TestTLSSysLogHandlerE2E(TestCase):
def setUp(self):
self.tmpdir = tempfile.TemporaryDirectory()
self.queue = multiprocessing.Queue(maxsize=1)
self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)
self._generate_keys()

def tearDown(self):
self.executor.shutdown(wait=True)
self.queue.close()
self.tmpdir.cleanup()

class TestTLSSysLogHandlerE2E(TestCertManager):
def _start_server_worker(self, sock_family, sock_type, sock_addr, secure):
if sock_type != socket.SOCK_DGRAM and sock_type != socket.SOCK_STREAM:
raise ValueError(
Expand Down Expand Up @@ -96,105 +68,6 @@ def _start_server(self, sock_family, sock_type, sock_addr, secure=False):
)
sleep(4)

# https://gist.github.com/bloodearnest/9017111a313777b9cce5
# Copyright 2018 Simon Davy
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
def _generate_selfsigned_cert(self, hostname, ip_addresses=None, key=None):
"""Generates self signed certificate for a hostname, and optional IP addresses."""
# Generate our key
if key is None:
key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend(),
)

name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, hostname)])

# best practice seem to be to include the hostname in the SAN, which *SHOULD* mean COMMON_NAME is ignored.
alt_names = [x509.DNSName(hostname)]

# allow addressing by IP, for when you don't have real DNS (common in most testing scenarios
if ip_addresses:
for addr in ip_addresses:
# openssl wants DNSnames for ips...
alt_names.append(x509.DNSName(addr))
# ... whereas golang's crypto/tls is stricter, and needs IPAddresses
# note: older versions of cryptography do not understand ip_address objects
alt_names.append(x509.IPAddress(ipaddress.ip_address(addr)))

san = x509.SubjectAlternativeName(alt_names)

# path_len=0 means this cert can only sign itself, not other certs.
basic_contraints = x509.BasicConstraints(ca=True, path_length=0)
now = datetime.datetime.now()
cert = (
x509.CertificateBuilder()
.subject_name(name)
.issuer_name(name)
.public_key(key.public_key())
.serial_number(1000)
.not_valid_before(now)
.not_valid_after(now + datetime.timedelta(days=10 * 365))
.add_extension(basic_contraints, False)
.add_extension(san, False)
.sign(key, hashes.SHA256(), default_backend())
)
cert_pem = cert.public_bytes(encoding=serialization.Encoding.PEM)
key_pem = key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
)

return cert_pem, key_pem

def _generate_keys(self):
pub_key_bytes, priv_key_bytes = self._generate_selfsigned_cert(
"localhost", ["::1", "127.0.0.1"]
)

pub_key_path = os.path.join(self.tmpdir.name, "syslog.pub")
priv_key_path = os.path.join(self.tmpdir.name, "syslog.key")

with open(pub_key_path, "wb") as f:
f.write(pub_key_bytes)

with open(priv_key_path, "wb") as f:
f.write(priv_key_bytes)

self.priv_key = priv_key_path
self.pub_key = pub_key_path

def _build_logger(self) -> logging.Logger:
stack = inspect.stack()
logger_name = "{}.{}".format(__name__, stack[1][3])

test_logger = logging.getLogger(logger_name)
test_logger.setLevel(logging.DEBUG)

for handler in test_logger.handlers:
test_logger.removeHandler(handler)

return test_logger

def test_e2e_unix_DGRAM(self):
self.socket_path = os.path.join(self.tmpdir.name, "syslog.sock")
self._start_server(socket.AF_UNIX, socket.SOCK_DGRAM, (self.socket_path,))
Expand Down
Loading