@@ 5,7 5,6 @@ import os
import time
from base64 import urlsafe_b64encode
from datetime import datetime, timezone
-from email.utils import formatdate
from json import dumps
from random import randint # noqa: DUO102
from socket import AI_CANONNAME, gaierror, getaddrinfo, gethostname
@@ 63,13 62,15 @@ def login_api(cnf):
"""
raise_unless_has_cnf(cnf)
- url = "{}/sessions".format(cnf.api)
+ challenge = get_signing_challenge(cnf)["data"][0]
+ challenge_id = challenge["id"]
+ signature = sign_data(cnf, challenge["attributes"]["value"].encode("utf-8"))
- now = time.time()
- sig = sign_data(cnf, url, now)
+ url = f"{cnf.api}/sessions"
headers = {
- "authorization": "X-Custos user=^{}, signature={}".format(cnf.agent, sig),
- "date": formatdate(timeval=now, usegmt=True),
+ "authorization": f"X-Custos user=^{cnf.agent}"
+ f", challenge=^{challenge_id}"
+ f", signature={signature}",
}
response = requests.post(url, headers=headers, timeout=API_TIMEOUT)
@@ 82,7 83,7 @@ def login_api(cnf):
session = response.json()["data"][0]
cnf.session = {
"token": session["attributes"]["token"],
- "expiration": now + session["meta"]["expiration_timeout"] - jitter,
+ "expiration": time.time() + session["meta"]["expiration_timeout"] - jitter,
}
@@ 200,20 201,34 @@ def build_authn_header(cnf):
return "X-Custos user=^{}, session=^{}".format(cnf.agent, cnf.session["token"])
-def sign_data(cnf, data, now):
+def get_signing_challenge(cnf):
+ """Gets the api's current authn challenge data.
+
+ Arguments:
+ cnf (Config): Config object.
+
+ Returns:
+ Response: Response json.
+ """
+ raise_unless_has_cnf(cnf)
+
+ response = requests.get(f"{cnf.api}/sessions/challenge", timeout=API_TIMEOUT)
+ response.raise_for_status()
+ return response.json()
+
+
+def sign_data(cnf, data):
"""Signs the specified message.
Arguments:
cnf (Config): Config object.
- data (str): Message to sign.
- now (float): Current time in seconds.
+ data (bytes): Message to sign.
Returns:
str: Base64-web-encoded signature.
"""
signing_key = load_signing_key(cnf)
- message = format_signed_message(data, now)
- signed_message = signing_key.sign(message)
+ signed_message = signing_key.sign(data)
return encode_base64_web(signed_message.signature)
@@ 244,19 259,6 @@ def load_signing_key(cnf):
_raise_setup_issue("Invalid private key in credentials file")
-def format_signed_message(data, now):
- """Formats the specified message with timestamp for signing.
-
- Arguments:
- data (str): Message to sign.
- now (float): Current time in seconds.
-
- Returns:
- bytes: Bytes ready to sign.
- """
- return (data + "#" + format_datetime(now)).encode("utf-8")
-
-
def encode_base64_web(message):
"""Base64-web-encodes the specified message.
@@ 7,7 7,6 @@ from unittest.mock import Mock
import pytest
from nacl.encoding import Base64Encoder
-from nacl.exceptions import BadSignatureError
from nacl.signing import SigningKey, VerifyKey
from procustodibus_agent import __version__ as version
@@ 15,9 14,9 @@ from procustodibus_agent.api import (
build_agent_info,
encode_base64_web,
format_datetime,
- format_signed_message,
get_health_info,
get_host_info,
+ get_signing_challenge,
getfqdn,
load_setup,
load_signing_key,
@@ 96,6 95,10 @@ def test_login_api_with_empty_interfaces(mocker):
response.status_code = 200
response.json = Mock(return_value=_session_json())
+ mocker.patch(
+ "procustodibus_agent.api.get_signing_challenge",
+ return_value=_challenge_json(),
+ )
mocker.patch("procustodibus_agent.api.randint", return_value=60)
mocker.patch("time.time", return_value=1500000000)
post = mocker.patch("requests.post", return_value=response)
@@ 106,10 109,9 @@ def test_login_api_with_empty_interfaces(mocker):
post.assert_called_once_with(
"http://localhost:8123/sessions",
headers={
- "authorization": "X-Custos user=^agent123, signature="
- "t02SiS8G8kI5YilAxJvbDTNA6OhHapym-FkRGuX_jtY"
- "pUMQJegQHIjv1Lp6U8y4xl0cVmWpi0a6icXmzQlTgAw",
- "date": "Fri, 14 Jul 2017 02:40:00 GMT",
+ "authorization": "X-Custos user=^agent123, challenge=^c123, signature="
+ "llNxBWHDFpt6lXegGVUWne8YP7OuKC4FvsYkgm4lWww"
+ "-7ePs_gVPtaQO_q7wQK-qRSIMzXv4QTulMfJPP4aSCQ",
},
timeout=16,
)
@@ 145,6 147,10 @@ def test_ping_api_authn_error(mocker):
login_response.status_code = 200
login_response.json = Mock(return_value=_session_json())
+ mocker.patch(
+ "procustodibus_agent.api.get_signing_challenge",
+ return_value=_challenge_json(),
+ )
mocker.patch("procustodibus_agent.api.randint", return_value=60)
mocker.patch("time.time", return_value=1500000000)
login_post = mocker.patch("requests.post", return_value=login_response)
@@ 166,6 172,10 @@ def test_ping_api_with_no_session(mocker):
login_response.status_code = 200
login_response.json = Mock(return_value=_session_json())
+ mocker.patch(
+ "procustodibus_agent.api.get_signing_challenge",
+ return_value=_challenge_json(),
+ )
mocker.patch("procustodibus_agent.api.randint", return_value=60)
mocker.patch("time.time", return_value=1500000000)
login_post = mocker.patch("requests.post", return_value=login_response)
@@ 206,6 216,17 @@ def test_host_info_with_good_response(mocker):
assert send.call_args[0][0].url == "http://localhost:8123/hosts/host123"
+def test_get_signing_challenge_with_good_response(mocker):
+ response = Mock()
+ response.status_code = 200
+ response.json = Mock(return_value="some json")
+ get = mocker.patch("requests.get", return_value=response)
+
+ assert get_signing_challenge(_basic_cnf()) == "some json"
+
+ get.assert_called_once_with("http://localhost:8123/sessions/challenge", timeout=16)
+
+
def test_load_signing_key_with_no_credentials():
cnf = _basic_cnf()
cnf.credentials = ""
@@ 246,30 267,31 @@ def test_load_signing_key_with_good_credentials(datadir):
@pytest.mark.parametrize(
- "data, sig",
+ "message, signature",
[
(
- "",
+ "test",
(
- "AZv6-arIGYyrNlayxR6gskktHYmPstul0MZcs2ZbI11"
- "thtGbPRzdzFKNSA1KfUM61HEO3HsyO2zwTE0uKncGDQ"
+ "llNxBWHDFpt6lXegGVUWne8YP7OuKC4FvsYkgm4lWww"
+ "-7ePs_gVPtaQO_q7wQK-qRSIMzXv4QTulMfJPP4aSCQ"
),
),
(
- "http://localhost:8123/hosts/host123/ping",
+ "PpmM3ixAJfItV8FapPYN",
(
- "GwFtnhpSzC9HACf5qpSweB4xGqXaF-UD4csYPceo86N"
- "YxRvPwuJL6iNhPWjkpbO3dzfQFF-2LjFrGfL8zHzEDw"
+ "QE8kZCNd3-fAfL6WQQFHUtAOT8asvUy3Omk5gvFDCuT"
+ "MlMnAR-2N2817X4g2EuugvucAOBA0KAbUEr8Y8kJACw"
),
),
],
)
-def test_sign_data(data, sig):
- assert sign_data(_basic_cnf(), data, 1500000000) == sig
+def test_sign_data(message, signature):
+ data = message.encode("utf-8")
+ cnf = _basic_cnf()
+
+ assert sign_data(cnf, data) == signature
- _verify_data(_basic_cnf(), data, 1500000000, sig)
- with pytest.raises(BadSignatureError):
- _verify_data(_basic_cnf(), data, 1500000001, sig)
+ _verify_signature(cnf, data, signature)
@pytest.mark.parametrize(
@@ 456,11 478,22 @@ def _session_json():
}
-def _verify_data(cnf, data, now, sig):
- message = format_signed_message(data, now)
- sig_bytes = urlsafe_b64decode(_pad_base64(sig))
+def _challenge_json():
+ return {
+ "data": [
+ {
+ "id": "c123",
+ "type": "system_authn_challenges",
+ "attributes": {"value": "test"},
+ }
+ ]
+ }
+
+
+def _verify_signature(cnf, data, signature):
+ signature_bytes = urlsafe_b64decode(_pad_base64(signature))
verify_key = VerifyKey(cnf.credentials["public_key"], encoder=Base64Encoder)
- verify_key.verify(message, sig_bytes)
+ verify_key.verify(data, signature_bytes)
def _pad_base64(s):