Skip to content

HMAC Helpers

Module: s2auth.common.hmac

CHARS = string.ascii_lowercase + string.ascii_uppercase + string.digits module-attribute

_ALGORITHM_MAP = OrderedDict([(HmacHashingAlgorithm.SHA256, hashlib.sha256)]) module-attribute

LOGGER = logging.getLogger(__name__) module-attribute

PairingToken = Annotated[str, StringConstraints(pattern='^[A-Za-z0-9+/]{4,}={0,2}$', min_length=4)] module-attribute

_ALL_UTF8_CHARS = [(chr(i)) for i in (range(1114112)) if not 55296 <= i <= 57343] module-attribute

IncompatibleHmacHashingAlgorithms

Bases: S2ConnectError

Raised when no common HMAC hashing algorithm is found

Source code in src/s2auth/common/exceptions.py 
41
42
43
44
class IncompatibleHmacHashingAlgorithms(S2ConnectError):
    """Raised when no common HMAC hashing algorithm is found"""

    error_type = PairingErrorType.IncompatibleHmacHashingAlgorithms

VerificationError

Bases: S2ConnectError

Digest verification failed

Source code in src/s2auth/common/exceptions.py 
47
48
49
50
class VerificationError(S2ConnectError):
    """Digest verification failed"""

    error_type = PairingErrorType.Other  # TODO should be something else.

HmacChallenge

Bases: RootModel[Base64Bytes]

Source code in src/s2auth/common/model/s2_connect_pairing.py 
62
63
64
65
66
67
68
class HmacChallenge(RootModel[Base64Bytes]):
    root: Annotated[
        Base64Bytes,
        Field(
            description="Random generated binary data encoded using Base64, used as the challenge for the HMAC based challenge response process. The challenge should be send to the other node as part of the pairing process. It must have a length of at least 32 bytes."
        ),
    ]

HmacHashingAlgorithm

Bases: str, Enum

Source code in src/s2auth/common/model/s2_connect_pairing.py 
58
59
class HmacHashingAlgorithm(str, Enum):
    SHA256 = "SHA256"

AccessToken

Bases: RootModel[Base64Bytes]

Source code in src/s2auth/common/model/s2_connect_common.py 
21
22
23
24
25
26
27
class AccessToken(RootModel[Base64Bytes]):
    root: Annotated[
        Base64Bytes,
        Field(
            description="One-time access token for secure access to the S2 message communication channel. It must be renewed every time a client wants to access the S2 message communication channel by calling the requestToken endpoint.This token is valid for one time login, with a maximum 5 years, and should have a minimum length of 32 bytes."
        ),
    ]

Deployment

Bases: str, Enum

Source code in src/s2auth/common/model/s2_connect_common.py 
44
45
46
class Deployment(str, Enum):
    WAN = "WAN"
    LAN = "LAN"

_get_hashing_algorithm(algorithm)

Source code in src/s2auth/common/hmac.py 
35
36
37
38
39
40
41
def _get_hashing_algorithm(algorithm: HmacHashingAlgorithm) -> Callable[..., Any]:
    try:
        return _ALGORITHM_MAP[algorithm]
    except KeyError as e:
        raise ValueError(
            f"Hashing algorithm '{algorithm}' is not supported. Please use one of {get_supported_algorithms()}"
        ) from e

create_pairing_code(s2_node_id=None, length=9)

Create pairing code, which is [pairing S2 node ID]-[pairing token] if the S2 node id is set otherwise just the token

Source code in src/s2auth/common/hmac.py 
51
52
53
54
55
56
57
58
59
60
61
62
@register_provider()
def create_pairing_code(s2_node_id: str | None = None, length: int = 9) -> PairingToken:
    """
    Create pairing code, which is [pairing S2 node ID]-[pairing token] if the S2 node id is set otherwise just the token
    """
    if length < 9:
        raise ValueError("The pairing token needs to be at least 9 bytes.")
    token_str = ''.join(random.choice(CHARS) for _ in range(length))

    if s2_node_id:
        return f"{s2_node_id}-{token_str}"
    return token_str

create_challenge(length=128)

Create the base64 encoded challenge (sequence of random bytes) to be sent to the other side of the connection. The challenge needs to be passed to the the other side of the connection, who should sign it with a shared pairing token. verify_response can then be used to verify that signature.

Source code in src/s2auth/common/hmac.py 
65
66
67
68
69
70
71
72
def create_challenge(length: int = 128) -> HmacChallenge:
    """
    Create the base64 encoded challenge (sequence of random bytes) to be sent to the other side of the connection.
    The challenge needs to be passed to the the other side of the connection, who should sign it with a shared pairing token.
    verify_response can then be used to verify that signature.
    """
    challenge_value: bytes = secrets.token_bytes(length)
    return HmacChallenge(root=b64encode(challenge_value))

generate_access_token()

Generate a cryptographically secure random access token.

Source code in src/s2auth/common/hmac.py 
75
76
77
78
@register_provider()
def generate_access_token() -> AccessToken:
    """Generate a cryptographically secure random access token."""
    return AccessToken(root=b64encode(secrets.token_bytes(32)))

get_supported_algorithms()

Get a list of supported HMAC algorithms.

Source code in src/s2auth/common/hmac.py 
81
82
83
def get_supported_algorithms() -> list[HmacHashingAlgorithm]:
    """Get a list of supported HMAC algorithms."""
    return list(_ALGORITHM_MAP.keys())

select_algorithm(node_algorithms)

Source code in src/s2auth/common/hmac.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def select_algorithm(
    node_algorithms: list[HmacHashingAlgorithm],
) -> HmacHashingAlgorithm:
    supported_algorithms = get_supported_algorithms()
    common = set(supported_algorithms) & set(node_algorithms)

    if not common:
        raise IncompatibleHmacHashingAlgorithms(
            f"Node does not support any of our algorithms: {get_supported_algorithms()}"
        )

    # Return the last algorithm from supported_algorithms that's in common
    # (later algorithms are preferred as they are typically stronger)
    return [alg for alg in supported_algorithms if alg in common][-1]

calculate_certificate_fingerprint(cert_der)

Return the SHA-256 fingerprint for a DER-encoded certificate.

Source code in src/s2auth/common/hmac.py 
102
103
104
def calculate_certificate_fingerprint(cert_der: bytes) -> bytes:
    """Return the SHA-256 fingerprint for a DER-encoded certificate."""
    return hashlib.sha256(cert_der).digest()

_leaf_certificate_bytes(certificate_bytes)

Return leaf certificate bytes from a cert file payload.

For PEM chain files this extracts and converts the first certificate block (the leaf) to DER bytes. For non-PEM inputs, bytes are returned unchanged.

Source code in src/s2auth/common/hmac.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def _leaf_certificate_bytes(certificate_bytes: bytes) -> bytes:
    """Return leaf certificate bytes from a cert file payload.

    For PEM chain files this extracts and converts the first certificate block
    (the leaf) to DER bytes. For non-PEM inputs, bytes are returned unchanged.
    """
    pem_begin = b"-----BEGIN CERTIFICATE-----"
    pem_end = b"-----END CERTIFICATE-----"

    begin_index = certificate_bytes.find(pem_begin)
    if begin_index == -1:
        return certificate_bytes

    end_index = certificate_bytes.find(pem_end, begin_index)
    if end_index == -1:
        raise ValueError("Malformed PEM certificate file: missing END CERTIFICATE marker.")
    end_index += len(pem_end)

    leaf_pem_bytes = certificate_bytes[begin_index:end_index]
    try:
        leaf_pem_text = leaf_pem_bytes.decode("ascii")
    except UnicodeDecodeError as exc:
        raise ValueError("Malformed PEM certificate file: non-ASCII certificate block.") from exc

    return ssl.PEM_cert_to_DER_cert(leaf_pem_text)

calculate_certificate_fingerprint_from_certificate_file(certificate_file)

Read a certificate file and return the SHA-256 fingerprint of the leaf certificate.

Source code in src/s2auth/common/hmac.py 
134
135
136
137
138
139
def calculate_certificate_fingerprint_from_certificate_file(
    certificate_file: str | Path,
) -> bytes:
    """Read a certificate file and return the SHA-256 fingerprint of the leaf certificate."""
    cert_bytes = Path(certificate_file).read_bytes()
    return calculate_certificate_fingerprint(_leaf_certificate_bytes(cert_bytes))

calculate_fingerprint_from_response_certificate(response)

Source code in src/s2auth/common/hmac.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def calculate_fingerprint_from_response_certificate(response: httpx.Response) -> bytes | None:
    network_stream = response.extensions.get("network_stream")
    if network_stream is None:
        return None

    get_extra_info = getattr(network_stream, "get_extra_info", None)
    if not callable(get_extra_info):
        return None

    ssl_object = get_extra_info("ssl_object")
    if ssl_object is None:
        return None

    get_peer_cert = getattr(ssl_object, "getpeercert", None)
    if not callable(get_peer_cert):
        return None

    cert_der = get_peer_cert(binary_form=True)
    if not isinstance(cert_der, (bytes, bytearray)) or not cert_der:
        return None

    return calculate_certificate_fingerprint(bytes(cert_der))

create_response(pairing_token, challenge, deployment, domain_name, fingerprint, algorithm=HmacHashingAlgorithm.SHA256)

Source code in src/s2auth/common/hmac.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def create_response(pairing_token: str,
                    challenge: HmacChallenge,
                    deployment: str | Deployment,
                    domain_name: str | None,
                    fingerprint: bytes | None,
                    algorithm: HmacHashingAlgorithm = HmacHashingAlgorithm.SHA256) -> bytes:
    try:
        digestmod = _get_hashing_algorithm(algorithm)
    except ValueError as e:
        raise VerificationError(str(e)) from e
    if deployment == Deployment.LAN:
        assert fingerprint is not None
        LOGGER.debug(f"Creating LAN HMAC response with fingerprint: fingerprint=CertificateHash(Sha256({list(fingerprint)}))")
        return hmac_response_lan(pairing_token.encode('utf-8'), challenge.root, fingerprint, digestmod)
    else:
        assert domain_name is not None
        LOGGER.debug(f"Creating WAN HMAC response with domain_name: {domain_name}")
        response =  hmac_response_wan(pairing_token.encode('utf-8'), challenge.root, domain_name, digestmod)
        LOGGER.debug(f"WAN HMAC response: {response}")
        return response

hmac_response_lan(pairing_token, challenge, fingerprint, digestmod)

Source code in src/s2auth/common/hmac.py 
187
188
189
190
191
192
193
194
def hmac_response_lan(pairing_token: bytes,
                      challenge: bytes,
                      fingerprint: bytes,
                      digestmod: Callable[..., Any]) -> bytes:
    assert fingerprint is not None, "fingerprint name missing"
    # Lan: R = HMAC(C, pairing_token || fingerprint)
    msg = pairing_token + fingerprint
    return hmac.new(key=challenge, msg=msg, digestmod=digestmod).digest()

hmac_response_wan(pairing_token, challenge, domain_name, digestmod)

Source code in src/s2auth/common/hmac.py 
196
197
198
199
200
201
202
203
def hmac_response_wan(pairing_token: bytes,
                      challenge: bytes,
                      domain_name: str,
                      digestmod: Callable[..., Any]) -> bytes:
    assert domain_name is not None, "Domain name missing"
    # Wan: R = HMAC(C, pairing_token || domain)
    msg = pairing_token + domain_name.encode("utf-8")
    return hmac.new(key=challenge, msg=msg, digestmod=digestmod).digest()

verify_response(pairing_token, challenge, response, deployment, domain_name, fingerprint, algorithm=HmacHashingAlgorithm.SHA256)

Verify that a received challenge response signature for correctness based on pairing token and algorithm.

Source code in src/s2auth/common/hmac.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def verify_response(
    pairing_token: str,
    challenge: HmacChallenge,
    response: bytes,
    deployment: str | Deployment,
    domain_name: str | None,
    fingerprint: bytes | None,
    algorithm: HmacHashingAlgorithm = HmacHashingAlgorithm.SHA256,
) -> bool:
    """
    Verify that a received challenge response signature for correctness based on pairing token and algorithm.
    """
    LOGGER.debug(f"pairing_token: {pairing_token}")
    LOGGER.debug(f"challenge: {challenge.root}")
    LOGGER.debug(f"challenge (as str): {b64encode(challenge.root)}")
    LOGGER.debug(f"response: {response}")
    LOGGER.debug(f"algorithm: {algorithm}")

    correct_digest = create_response(pairing_token, challenge, deployment, domain_name, fingerprint, algorithm)
    LOGGER.debug(f"expected response: {correct_digest}")
    if not hmac.compare_digest(correct_digest, response):
        raise VerificationError("Signature is invalid.")
    return True