Add bucket notifications test

Author: href

README.md

@@ -46,6 +46,7 @@ These tests are run regularly against our public infrastructure as well as our i
 | **Nested Virtualization** | [test_virtualization_support](./test_nested_virtualization.py#L14)               | default  |
 |                           | [test_run_nested_vm](./test_nested_virtualization.py#L40)                        | default  |
 | **Objects**               | [test_bucket_urls](./test_objects.py#L23)                                        | default  |
+|                           | [test_notifications](./test_objects.py#L69)                                      | default  |
 | **Private Network**       | [test_private_ip_address_on_all_images](./test_private_network.py#L14)           | all      |
 |                           | [test_private_network_connectivity_on_all_images](./test_private_network.py#L35) | all      |
 |                           | [test_multiple_private_network_interfaces](./test_private_network.py#L88)        | default  |

api.py

@@ -279,4 +279,17 @@ def delete_objects_users(api, url):
         bucket.objects.all().delete()
         bucket.delete()
 
+    sns = boto3.client(
+        'sns',
+        endpoint_url=objects_endpoint,
+        aws_access_key_id=user.keys[0]['access_key'],
+        aws_secret_access_key=user.keys[0]['secret_key'],
+        region_name='default',
+    )
+
+    for topic in sns.list_topics().get('Topics', ()):
+        arn = topic["TopicArn"]
+        assert re.match(r'arn:aws:sns:(rma|lpg)::at-.+', arn)
+        sns.delete_topic(TopicArn=arn)
+
     api.request("DELETE", url)

resources.py

@@ -980,6 +980,11 @@ def put_file_content(self, remote_filename, content, sudo=False):
 
             self.put_file(f.name, remote_filename, sudo)
 
+    def get_file_handle(self, remote_filename):
+
+        sftp = self.host.backend.client.open_sftp()
+        return sftp.open(remote_filename)
+
     def enable_dhcp_in_networkd(self, interface):
         """ Additional private network interfaces have to be explicitly
         configured to use DHCP, to get an IP address.

scripts/sns-endpoint-test-server

@@ -0,0 +1,67 @@
+#!/usr/bin/env -S python3 -u
+# Note: It's important to use unbuffered output, otherwise log
+# lines won't appear in the journal immediately.
+""" HTTP test server for object storage notifications tests
+
+This server acts as the endpoint for the Simple Notification Service (SNS).
+
+This is based on the HTTP server included in the http.server standard library
+module. It has the following features:
+* Only responds to POST requests. All other requests are not handled. No access
+  to the filesystem is possible.
+* The body of the request is logged to a separate log file.
+"""
+
+import http.server
+import ssl
+
+from argparse import ArgumentParser
+from http import HTTPStatus
+from pathlib import Path
+
+
+class SNSEndpointRequestHandler(http.server.BaseHTTPRequestHandler):
+
+    server_version = f'SNSEndpointTestHTTP/{http.server.__version__}'
+    protocol_version = 'HTTP/1.1'
+
+    def do_POST(self):
+
+        # Log some details about the notification record
+        length = int(self.headers.get('content-length', 0))
+        if length:
+            body = self.rfile.read(length)
+
+            with body_log_file.open('ab') as body_log:
+                body_log.write(body + b'\n')
+
+        self.send_response(HTTPStatus.NO_CONTENT)
+        self.end_headers()
+
+
+parser = ArgumentParser()
+parser.add_argument('--ssl', action='store_true')
+parser.add_argument('--port', type=int, default=8000)
+parser.add_argument('--body-log', default='notification-body.log')
+
+args = parser.parse_args()
+
+with http.server.ThreadingHTTPServer(
+        ('', args.port),
+        SNSEndpointRequestHandler
+) as server:
+    print(f'SNS Endpoint test HTTP server running on '
+          f'{":".join(map(str, server.server_address))}')
+    if args.ssl:
+        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+        context.load_cert_chain('./server.pem')
+        server.socket = context.wrap_socket(server.socket, server_side=True)
+
+    body_log_file = Path(args.body_log)
+
+    # Create an empty log file
+    if body_log_file.exists():
+        body_log_file.unlink()
+    body_log_file.touch()
+
+    server.serve_forever()

test_objects.py

@@ -9,8 +9,13 @@
 """
 
 import boto3
+import json
 import requests
 import secrets
+import time
+
+from util import flatten
+from util import setup_notification_endpoint
 
 from urllib.parse import urlparse
 
@@ -59,3 +64,90 @@ def test_bucket_urls(objects_endpoint, access_key, secret_key):
         response = requests.get(url)
         assert response.status_code == 200
         assert response.text == "test"
+
+
+def test_notifications(
+    bucket,
+    access_key,
+    secret_key,
+    objects_endpoint,
+    server,
+    region,
+):
+    """ Using S3 SNS (Simple Notification Service) we can be informed via
+    webhooks, when something changes on a bucket.
+
+    """
+
+    # Run a service that can act as a webhook endpoint
+    setup_notification_endpoint(server)
+
+    # Get an SNS client (Simple Notification Service)
+    sns = boto3.client(
+        'sns',
+        endpoint_url=objects_endpoint,
+        aws_access_key_id=access_key,
+        aws_secret_access_key=secret_key,
+        region_name='default',
+    )
+
+    # Create a test topic
+    name = f"at-{secrets.token_hex(8)}"
+
+    topic = sns.create_topic(Name=name, Attributes={
+        "push-endpoint": f"http://{server.ip('public', 4)}:8000",
+    })
+
+    # Get notified whenever an object is created
+    bucket.Notification().put(NotificationConfiguration={
+        "TopicConfigurations": [
+            {
+                "Id": name,
+                "TopicArn": topic['TopicArn'],
+                "Events": ["s3:ObjectCreated:*"]
+            }
+        ]
+    })
+
+    # We have to wait a moment for the configuration to propagate
+    timeout = time.monotonic() + 30
+
+    while time.monotonic() < timeout:
+        bucket.put_object(Key='pre-check', Body=b'')
+
+        found = False
+        with server.get_file_handle('notification-body.log') as notifications:
+            for line in notifications:
+                n = json.loads(line)
+                for r in n['Records']:
+                    if r['s3']['object']['key'] == 'pre-check':
+                        found = True
+
+        if found:
+            break
+
+        time.sleep(1)
+
+    # Generate multiple notifications
+    for i in range(3):
+        bucket.put_object(Key=f'count-{i}', Body=b'')
+
+    # Ensure they were received (excluding the pre-check objects from above)
+    with server.get_file_handle('notification-body.log') as notification_log:
+        notifications = [json.loads(line) for line in notification_log
+                         if 'pre-check' not in line]
+
+    # A single message may contain multiple records
+    records = flatten(m['Records'] for m in notifications)
+    assert len(records) == 3
+
+    # The records are sent in order
+    assert records[0]['s3']['object']['key'] == 'count-0'
+    assert records[1]['s3']['object']['key'] == 'count-1'
+    assert records[2]['s3']['object']['key'] == 'count-2'
+
+    # They all share some properties
+    for record in records:
+        assert record['eventName'] == 'ObjectCreated:Put'
+        assert record['awsRegion'] == region
+        assert record['s3']['bucket']['name'] == bucket.name

util.py

@@ -26,6 +26,7 @@
 from hashlib import blake2b
 from ipaddress import ip_address
 from ipaddress import ip_network
+from itertools import chain
 from paramiko import SSHClient, AutoAddPolicy
 from paramiko.ssh_exception import ChannelException
 from paramiko.ssh_exception import NoValidConnectionsError
@@ -733,6 +734,25 @@ def setup_lbaas_backend(backend, backend_network, ssl=False, protocol='tcp'):
         raise AssertionError(f'Unknown protocol: {protocol}')
 
 
+def setup_notification_endpoint(endpoint):
+    """ Configures a server to work as a S3 SNS notification endpoint.
+
+    A simple HTTP test server is started to serve a webhook URL for a S3 SNS
+    notification endpoint.
+    """
+
+    # Copy test server Python script to backend server
+    endpoint.put_file('scripts/sns-endpoint-test-server')
+    endpoint.run('chmod +x sns-endpoint-test-server')
+
+    endpoint.run(oneliner(f'''
+        systemd-run
+        --user
+        --unit sns-endpoint
+        ./sns-endpoint-test-server
+    '''))
+
+
 def wait_for_url_ready(url, prober, content=None, timeout=90):
     """ Waits for an URL to return an OK status code or specific content. """
 
@@ -856,3 +876,9 @@ def skip_test_when(match, reason=None):
             pytest.skip(reason)
 
         raise
+
+
+def flatten(list_of_lists):
+    """ Flatten one level of nesting. """
+
+    return list(chain.from_iterable(list_of_lists))