stack/tests/api/test_sigv4.py

"""Tests for AWS SigV4 signing — api.clients.rustfs.sigv4."""

from __future__ import annotations

import hashlib
import hmac
from datetime import datetime, timezone

from api.clients.rustfs.sigv4 import (
    _canonical_query_string,
    _get_signature_key,
    _sign,
    sign_request,
)


class TestSign:
    def test_returns_bytes(self) -> None:
        result = _sign(b"key", "msg")
        assert isinstance(result, bytes)
        assert len(result) == 32  # SHA-256 digest is 32 bytes

    def test_deterministic(self) -> None:
        a = _sign(b"key", "hello")
        b = _sign(b"key", "hello")
        assert a == b

    def test_different_msgs(self) -> None:
        a = _sign(b"key", "hello")
        b = _sign(b"key", "world")
        assert a != b

    def test_different_keys(self) -> None:
        a = _sign(b"key1", "msg")
        b = _sign(b"key2", "msg")
        assert a != b

    def test_matches_stdlib(self) -> None:
        key = b"test-key"
        msg = "test-message"
        expected = hmac.new(key, msg.encode(), hashlib.sha256).digest()
        assert _sign(key, msg) == expected


class TestGetSignatureKey:
    def test_returns_bytes(self) -> None:
        key = _get_signature_key("mysecret", "20260101", "us-east-1", "s3")
        assert isinstance(key, bytes)

    def test_deterministic(self) -> None:
        a = _get_signature_key("secret", "20260101", "us-east-1", "s3")
        b = _get_signature_key("secret", "20260101", "us-east-1", "s3")
        assert a == b

    def test_different_regions_produce_different_keys(self) -> None:
        a = _get_signature_key("secret", "20260101", "us-east-1", "s3")
        b = _get_signature_key("secret", "20260101", "eu-west-1", "s3")
        assert a != b

    def test_different_dates_produce_different_keys(self) -> None:
        a = _get_signature_key("secret", "20260101", "us-east-1", "s3")
        b = _get_signature_key("secret", "20260102", "us-east-1", "s3")
        assert a != b

    def test_different_services_produce_different_keys(self) -> None:
        a = _get_signature_key("secret", "20260101", "us-east-1", "s3")
        b = _get_signature_key("secret", "20260101", "us-east-1", "ec2")
        assert a != b


class TestCanonicalQueryString:
    def test_empty_params(self) -> None:
        assert _canonical_query_string(None) == ""
        assert _canonical_query_string({}) == ""

    def test_single_param(self) -> None:
        result = _canonical_query_string({"list-type": "2"})
        assert result == "list-type=2"

    def test_params_sorted(self) -> None:
        result = _canonical_query_string({"z": "1", "a": "2"})
        assert result == "a=2&z=1"

    def test_special_chars_encoded(self) -> None:
        result = _canonical_query_string({"key": "hello world"})
        assert result == "key=hello%20world"

    def test_slash_encoded(self) -> None:
        result = _canonical_query_string({"prefix": "data/files/"})
        assert "data%2Ffiles%2F" in result

    def test_multiple_params(self) -> None:
        result = _canonical_query_string({"b": "2", "a": "1", "c": "3"})
        assert result == "a=1&b=2&c=3"


class TestSignRequest:
    _NOW = datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc)

    def _sign(self, method="GET", url="http://s3.example.com/bucket/key", **kwargs):
        return sign_request(
            method=method,
            url=url,
            headers={},
            body=None,
            access_key="AKIAIOSFODNN7EXAMPLE",
            secret_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
            now=self._NOW,
            **kwargs,
        )

    def test_returns_dict(self) -> None:
        result = self._sign()
        assert isinstance(result, dict)

    def test_authorization_header_present(self) -> None:
        result = self._sign()
        assert "Authorization" in result

    def test_authorization_starts_with_algo(self) -> None:
        result = self._sign()
        assert result["Authorization"].startswith("AWS4-HMAC-SHA256 ")

    def test_x_amz_date_set(self) -> None:
        result = self._sign()
        assert result["x-amz-date"] == "20260115T120000Z"

    def test_x_amz_content_sha256_set(self) -> None:
        result = self._sign()
        assert "x-amz-content-sha256" in result

    def test_body_is_none_uses_empty_hash(self) -> None:
        result = self._sign()  # body=None by default via _sign helper
        # SHA-256 of empty bytes (None treated as b"")
        empty_hash = hashlib.sha256(b"").hexdigest()
        assert result["x-amz-content-sha256"] == empty_hash

    def test_body_hash_included(self) -> None:
        result = sign_request(
            method="PUT",
            url="http://s3.example.com/bucket/key",
            headers={},
            body=b"hello world",
            access_key="AKID",
            secret_key="SECRET",
            now=self._NOW,
        )
        expected = hashlib.sha256(b"hello world").hexdigest()
        assert result["x-amz-content-sha256"] == expected

    def test_authorization_contains_credential(self) -> None:
        result = self._sign()
        assert "Credential=AKIAIOSFODNN7EXAMPLE/20260115" in result["Authorization"]

    def test_authorization_contains_signed_headers(self) -> None:
        result = self._sign()
        assert "SignedHeaders=" in result["Authorization"]

    def test_authorization_contains_signature(self) -> None:
        result = self._sign()
        assert "Signature=" in result["Authorization"]

    def test_non_standard_port_signed(self) -> None:
        """Port 9000 should be included in the host header (affects signature)."""
        result_with_port = self._sign(url="http://s3.example.com:9000/bucket/key")
        result_without_port = self._sign(url="http://s3.example.com/bucket/key")
        # Different hosts → different signatures
        assert result_with_port["Authorization"] != result_without_port["Authorization"]

    def test_standard_port_80_same_as_no_port(self) -> None:
        """Port 80 on HTTP is standard and should not affect the host."""
        result_port80 = self._sign(url="http://s3.example.com:80/bucket/key")
        result_no_port = self._sign(url="http://s3.example.com/bucket/key")
        assert result_port80["Authorization"] == result_no_port["Authorization"]

    def test_https_port_443_same_as_no_port(self) -> None:
        """Port 443 on HTTPS is standard and should not affect the host."""
        result_443 = self._sign(url="https://s3.example.com:443/bucket/key")
        result_no_port = self._sign(url="https://s3.example.com/bucket/key")
        assert result_443["Authorization"] == result_no_port["Authorization"]

    def test_existing_headers_preserved(self) -> None:
        # Use all-lowercase header to avoid the dead-code bug in lines 96-100
        # (the code overwrites canonical_headers at line 102 anyway)
        result = sign_request(
            method="GET",
            url="http://s3.example.com/",
            headers={"x-custom": "value"},
            body=None,
            access_key="AKID",
            secret_key="SECRET",
            now=self._NOW,
        )
        assert "x-custom" in result

    def test_query_string_params_signed(self) -> None:
        result = self._sign(
            url="http://s3.example.com/bucket?list-type=2&prefix=data%2F"
        )
        auth = result["Authorization"]
        assert "Signature=" in auth

    def test_now_defaults_to_utc(self) -> None:
        # When now=None, it uses current UTC time — just verify it doesn't crash
        result = sign_request(
            method="GET",
            url="http://s3.example.com/",
            headers={},
            body=None,
            access_key="AKID",
            secret_key="SECRET",
        )
        assert "Authorization" in result

    def test_custom_region_and_service(self) -> None:
        result = sign_request(
            method="GET",
            url="http://example.com/",
            headers={},
            body=None,
            access_key="AKID",
            secret_key="SECRET",
            region="eu-west-1",
            service="execute-api",
            now=self._NOW,
        )
        assert "eu-west-1/execute-api/aws4_request" in result["Authorization"]

    def test_deterministic_with_fixed_time(self) -> None:
        r1 = self._sign()
        r2 = self._sign()
        assert r1["Authorization"] == r2["Authorization"]