inkwell/tests/test_backup_s3.py

import threading
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import pytest

from inkwell.backup_s3 import EMPTY_SHA256, S3Error, S3Target, list_objects, put_file, sign

# AWS's published Signature Version 4 examples for S3 ("Signature Calculations for the
# Authorization Header: Transferring Payload in a Single Chunk").
AK = "AKIAIOSFODNN7EXAMPLE"
SK = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
AT = datetime(2013, 5, 24, tzinfo=timezone.utc)


def _signature(headers: dict) -> str:
    return headers["Authorization"].rsplit("Signature=", 1)[1]


def test_get_object_matches_aws_example():
    headers = sign(
        "GET", "https://examplebucket.s3.amazonaws.com/test.txt", {"Range": "bytes=0-9"},
        EMPTY_SHA256, AK, SK, "us-east-1", AT,
    )
    assert "SignedHeaders=host;range;x-amz-content-sha256;x-amz-date" in headers["Authorization"]
    assert _signature(headers) == "f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41"


def test_list_objects_matches_aws_example():
    headers = sign(
        "GET", "https://examplebucket.s3.amazonaws.com/?max-keys=2&prefix=J", {},
        EMPTY_SHA256, AK, SK, "us-east-1", AT,
    )
    assert _signature(headers) == "34b48302e7b5fa45bde8084f4b7868a86f0a534bc59db6670ed5711ef69dc6f7"


# ---- against a local stand-in bucket ---------------------------------------------------

LISTING = b"""<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
  <Contents><Key>inkwell-20261015T020000Z.tar</Key><Size>42</Size><LastModified>2026-10-15T02:00:05Z</LastModified></Contents>
</ListBucketResult>"""

DENIED = b"""<?xml version="1.0" encoding="UTF-8"?>
<Error><Code>AccessDenied</Code><Message>Access Denied</Message></Error>"""


@pytest.fixture
def bucket():
    received: dict = {"puts": {}, "deny": False}

    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass

        def do_PUT(self):
            length = int(self.headers["Content-Length"])
            received["puts"][self.path] = (self.rfile.read(length), dict(self.headers))
            if received["deny"]:
                self.send_response(403)
                self.end_headers()
                self.wfile.write(DENIED)
                return
            self.send_response(200)
            self.send_header("Content-Length", "0")
            self.end_headers()

        def do_GET(self):
            received["get"] = self.path
            self.send_response(200)
            self.send_header("Content-Length", str(len(LISTING)))
            self.end_headers()
            self.wfile.write(LISTING)

    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    received["target"] = S3Target(f"http://127.0.0.1:{server.server_address[1]}", "backups", AK, SK)
    yield received
    server.shutdown()


def test_a_file_is_put_whole_and_signed(bucket, tmp_path):
    path = tmp_path / "inkwell-x.tar"
    path.write_bytes(b"x" * 300_000)
    put_file(bucket["target"], "inkwell-x.tar", path, limit_kbps=100_000)
    body, headers = bucket["puts"]["/backups/inkwell-x.tar"]
    assert body == b"x" * 300_000
    assert headers["Authorization"].startswith(f"AWS4-HMAC-SHA256 Credential={AK}/")
    assert headers["Host"] == bucket["target"].endpoint.split("//", 1)[1]


def test_a_refusal_says_why_in_words(bucket, tmp_path):
    bucket["deny"] = True
    path = tmp_path / "inkwell-x.tar"
    path.write_bytes(b"x")
    with pytest.raises(S3Error, match="AccessDenied: the key isn't allowed to do that"):
        put_file(bucket["target"], "inkwell-x.tar", path)


def test_listing_reads_the_objects(bucket):
    objects = list_objects(bucket["target"], "inkwell-")
    assert [(o.key, o.size) for o in objects] == [("inkwell-20261015T020000Z.tar", 42)]
    assert "list-type=2" in bucket["get"] and "prefix=inkwell-" in bucket["get"]


def test_an_address_nothing_answers_on_says_so():
    target = S3Target("http://127.0.0.1:1", "backups", AK, SK)
    with pytest.raises(S3Error, match="Couldn't reach"):
        list_objects(target)