George Lambert
c547804835
signup calls POST /auth/signup and leaves the stored JWT in place. login writes the VeraeTime token to the auth properties file and does not return it. A stored username and password refresh that token when expires_at is missing or within five minutes. create_user still requires an admin session.
308 lines
11 KiB
Python
308 lines
11 KiB
Python
"""Global VeraeTime client: same wire shape as verae-timestamp/global.Client."""
|
|
|
|
import pytest
|
|
|
|
from verae_mcp_timestamping_server.global_client import GlobalClient, GlobalLedger
|
|
from verae_mcp_timestamping_server.ledger import SHA256_EMPTY, MemoryLedger, RoutingLedger
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, status_code: int, body: dict):
|
|
self.status_code = status_code
|
|
self.body = body
|
|
self.text = str(body)
|
|
self.headers = {"Content-Type": "application/json"}
|
|
|
|
def json(self):
|
|
return self.body
|
|
|
|
|
|
class FakeHttp:
|
|
def __init__(self):
|
|
self.calls: list[tuple[str, str, dict, dict]] = []
|
|
|
|
def request(self, method: str, url: str, payload: dict, headers: dict) -> _Resp:
|
|
self.calls.append((method, url, payload, headers))
|
|
if url.endswith("/api/timestamp"):
|
|
return _Resp(
|
|
200,
|
|
{
|
|
"jobId": "job-1",
|
|
"hash": payload.get("data") or payload.get("hash"),
|
|
"timestamp": "2026-08-15T00:00:00Z",
|
|
"blockHash": "0xdead",
|
|
},
|
|
)
|
|
if url.endswith("/api/verify"):
|
|
if payload.get("certificate") == "job-1":
|
|
return _Resp(200, {"valid": True, "hash": SHA256_EMPTY})
|
|
return _Resp(200, {"valid": False})
|
|
if "/api/status/" in url:
|
|
return _Resp(
|
|
200,
|
|
{
|
|
"id": "job-1",
|
|
"status": "completed",
|
|
"result": "job-1",
|
|
"metadata": {"timestamp": "2026-08-15T00:00:00Z", "blockHash": "0xdead"},
|
|
},
|
|
)
|
|
return _Resp(404, {"error": "nope"})
|
|
|
|
|
|
async def test_global_stamp_posts_hash_sync_and_maps_receipt():
|
|
http = FakeHttp()
|
|
client = GlobalClient("https://api.veraetime.net", token="secret", http=http)
|
|
ledger = GlobalLedger(client)
|
|
[rec] = await ledger.stamp(SHA256_EMPTY, scope="global", cube_id="cube-1")
|
|
method, url, payload, headers = http.calls[0]
|
|
assert method == "POST"
|
|
assert url == "https://api.veraetime.net/api/timestamp"
|
|
assert payload["data"] == SHA256_EMPTY
|
|
assert payload["digest"] == SHA256_EMPTY
|
|
assert payload["hashAlg"] == "SHA256"
|
|
assert payload["restamp"] is False
|
|
assert "wait" not in payload
|
|
assert "digest" in payload
|
|
assert headers["Authorization"] == "Bearer secret"
|
|
assert rec.scope == "global"
|
|
assert rec.receipt_id == "job-1"
|
|
assert rec.merkle == "0xdead"
|
|
assert rec.cube_id == "cube-1"
|
|
|
|
|
|
async def test_existing_digest_returns_first_certificate_without_status_poll():
|
|
class ExistingHttp(FakeHttp):
|
|
def request(self, method, url, payload, headers):
|
|
self.calls.append((method, url, payload, headers))
|
|
if url.endswith("/api/timestamp"):
|
|
return _Resp(
|
|
200,
|
|
{
|
|
"jobId": "28df9952-094b-45a9-b6e5-e3ff81389bdd",
|
|
"existing": True,
|
|
"certificate": "QmVBa3op7DHgxSDgMURFpfGxTQ6UnZvzLsBCtpmLVAjXbA",
|
|
},
|
|
)
|
|
raise AssertionError(f"unexpected {method} {url}")
|
|
|
|
http = ExistingHttp()
|
|
client = GlobalClient("https://api.veraetime.net", http=http)
|
|
rec = client.submit(SHA256_EMPTY)
|
|
assert rec.receipt_id == "QmVBa3op7DHgxSDgMURFpfGxTQ6UnZvzLsBCtpmLVAjXbA"
|
|
assert http.calls[0][2]["digest"] == SHA256_EMPTY
|
|
assert http.calls[0][2]["restamp"] is False
|
|
assert len(http.calls) == 1
|
|
|
|
|
|
async def test_existing_without_certificate_polls_job():
|
|
class PendingExisting(FakeHttp):
|
|
def request(self, method, url, payload, headers):
|
|
self.calls.append((method, url, payload, headers))
|
|
if url.endswith("/api/timestamp"):
|
|
return _Resp(200, {"jobId": "job-9", "existing": True})
|
|
if url.endswith("/api/status/job-9"):
|
|
return _Resp(
|
|
200,
|
|
{"id": "job-9", "status": "completed", "result": "QmPending"},
|
|
)
|
|
return _Resp(404, {"error": "nope"})
|
|
|
|
http = PendingExisting()
|
|
rec = GlobalClient("https://api.veraetime.net", http=http).submit(SHA256_EMPTY)
|
|
assert rec.receipt_id == "QmPending"
|
|
assert http.calls[1][0] == "GET"
|
|
|
|
|
|
async def test_global_verify_posts_certificate():
|
|
http = FakeHttp()
|
|
ledger = GlobalLedger(GlobalClient("https://api.veraetime.net", http=http))
|
|
ok = await ledger.verify("job-1", want_scope="global")
|
|
assert ok.valid is True
|
|
assert ok.receipt is not None
|
|
assert ok.receipt.hash == SHA256_EMPTY
|
|
assert ok.receipt.scope == "global"
|
|
_, url, payload, _ = http.calls[0]
|
|
assert url.endswith("/api/verify")
|
|
assert payload == {"certificate": "job-1"}
|
|
|
|
|
|
async def test_routing_sends_only_global_scope_to_central():
|
|
http = FakeHttp()
|
|
local = MemoryLedger()
|
|
routed = RoutingLedger(
|
|
local=local,
|
|
global_ledger=GlobalLedger(GlobalClient("https://api.veraetime.net", http=http)),
|
|
)
|
|
[local_rec] = await routed.stamp(SHA256_EMPTY, scope="local")
|
|
assert local_rec.scope == "local"
|
|
assert http.calls == []
|
|
|
|
[g] = await routed.stamp(SHA256_EMPTY, scope="global", cube_id="cube-1")
|
|
assert g.scope == "global"
|
|
assert g.receipt_id == "job-1"
|
|
assert http.calls[0][1].endswith("/api/timestamp")
|
|
|
|
cached = await routed.hash_report(SHA256_EMPTY, want_scope="global")
|
|
assert cached == []
|
|
|
|
weak = await routed.verify(local_rec.receipt_id, want_scope="global")
|
|
assert weak.valid is False
|
|
|
|
|
|
def _auth_http(role: str = "admin"):
|
|
class AuthHttp(FakeHttp):
|
|
def request(self, method, url, payload, headers):
|
|
self.calls.append((method, url, payload, headers))
|
|
if url.endswith("/auth/login"):
|
|
return _Resp(
|
|
200,
|
|
{
|
|
"token": "fresh-token",
|
|
"expiresAt": "2099-01-01T00:00:00Z",
|
|
"user": {"id": "u1", "username": "admin", "role": "admin"},
|
|
},
|
|
)
|
|
if url.endswith("/auth/validate"):
|
|
if headers.get("Authorization") != "Bearer fresh-token":
|
|
return _Resp(401, {"error": "Unauthorized"})
|
|
return _Resp(
|
|
200,
|
|
{
|
|
"valid": True,
|
|
"userId": "u1",
|
|
"username": "admin",
|
|
"role": role,
|
|
"expiresAt": "2099-01-01T00:00:00Z",
|
|
},
|
|
)
|
|
if url.endswith("/auth/signup"):
|
|
assert "Authorization" not in headers
|
|
return _Resp(
|
|
201,
|
|
{"id": "u3", "username": payload["username"], "role": "user"},
|
|
)
|
|
if url.endswith("/auth/users"):
|
|
return _Resp(
|
|
201,
|
|
{"id": "u2", "username": payload["username"], "role": payload["role"]},
|
|
)
|
|
if url.endswith("/api/timestamp"):
|
|
return _Resp(
|
|
200,
|
|
{
|
|
"jobId": "job-1",
|
|
"existing": True,
|
|
"certificate": "QmFresh",
|
|
},
|
|
)
|
|
return _Resp(404, {"error": "nope"})
|
|
|
|
return AuthHttp()
|
|
|
|
|
|
async def test_login_stores_token_and_hides_it_from_result(tmp_path):
|
|
http = _auth_http()
|
|
path = tmp_path / "auth.properties"
|
|
client = GlobalClient("https://api.veraetime.net", http=http, auth_path=path)
|
|
result = client.login("admin", "secret-pass")
|
|
assert result.username == "admin"
|
|
assert result.role == "admin"
|
|
assert result.expires_at == "2099-01-01T00:00:00Z"
|
|
assert "fresh-token" not in result.model_dump().values()
|
|
stored = path.read_text()
|
|
assert "access_token=fresh-token" in stored
|
|
assert "username=admin" in stored
|
|
assert "password=secret-pass" in stored
|
|
assert path.stat().st_mode & 0o777 == 0o600
|
|
assert client.token == "fresh-token"
|
|
assert http.calls[0][2] == {"username": "admin", "password": "secret-pass"}
|
|
|
|
|
|
async def test_session_and_create_user_require_admin(tmp_path):
|
|
http = _auth_http(role="user")
|
|
client = GlobalClient(
|
|
"https://api.veraetime.net",
|
|
token="fresh-token",
|
|
http=http,
|
|
auth_path=tmp_path / "auth.properties",
|
|
)
|
|
seen = client.session()
|
|
assert seen.valid is True
|
|
assert seen.role == "user"
|
|
with pytest.raises(ValueError, match="admin login required"):
|
|
client.create_user("newuser", "secret-pass", "user")
|
|
assert not any(url.endswith("/auth/users") for _, url, _, _ in http.calls)
|
|
|
|
|
|
async def test_create_user_posts_when_admin(tmp_path):
|
|
http = _auth_http(role="admin")
|
|
client = GlobalClient(
|
|
"https://api.veraetime.net",
|
|
token="fresh-token",
|
|
http=http,
|
|
auth_path=tmp_path / "auth.properties",
|
|
)
|
|
created = client.create_user("newuser", "secret-pass", "user")
|
|
assert created.id == "u2"
|
|
assert created.username == "newuser"
|
|
assert created.role == "user"
|
|
assert "secret-pass" not in created.model_dump().values()
|
|
method, url, payload, headers = http.calls[-1]
|
|
assert method == "POST"
|
|
assert url.endswith("/auth/users")
|
|
assert payload["password"] == "secret-pass"
|
|
assert headers["Authorization"] == "Bearer fresh-token"
|
|
|
|
|
|
async def test_signup_creates_user_without_replacing_token(tmp_path):
|
|
http = _auth_http()
|
|
client = GlobalClient(
|
|
"https://api.veraetime.net",
|
|
token="service-token",
|
|
http=http,
|
|
auth_path=tmp_path / "auth.properties",
|
|
)
|
|
created = client.signup("newcomer", "secret-pass")
|
|
assert created == type(created)(id="u3", username="newcomer", role="user")
|
|
assert client.token == "service-token"
|
|
method, url, payload, headers = http.calls[0]
|
|
assert method == "POST"
|
|
assert url.endswith("/auth/signup")
|
|
assert payload == {"username": "newcomer", "password": "secret-pass"}
|
|
assert "Authorization" not in headers
|
|
assert "role" not in payload
|
|
|
|
|
|
async def test_submit_refreshes_expired_stored_login(tmp_path):
|
|
path = tmp_path / "auth.properties"
|
|
path.write_text(
|
|
"access_token=old-token\n"
|
|
"expires_at=2020-01-01T00:00:00Z\n"
|
|
"username=admin\n"
|
|
"password=secret-pass\n"
|
|
)
|
|
http = _auth_http()
|
|
client = GlobalClient("https://api.veraetime.net", http=http, auth_path=path)
|
|
rec = client.submit(SHA256_EMPTY)
|
|
assert rec.receipt_id == "QmFresh"
|
|
assert http.calls[0][1].endswith("/auth/login")
|
|
assert http.calls[1][3]["Authorization"] == "Bearer fresh-token"
|
|
|
|
|
|
async def test_submit_skips_login_when_token_is_fresh(tmp_path):
|
|
path = tmp_path / "auth.properties"
|
|
path.write_text(
|
|
"access_token=still-good\n"
|
|
"expires_at=2099-01-01T00:00:00Z\n"
|
|
"username=admin\n"
|
|
"password=secret-pass\n"
|
|
)
|
|
http = _auth_http()
|
|
client = GlobalClient("https://api.veraetime.net", http=http, auth_path=path)
|
|
rec = client.submit(SHA256_EMPTY)
|
|
assert rec.receipt_id == "QmFresh"
|
|
assert http.calls[0][1].endswith("/api/timestamp")
|
|
assert http.calls[0][3]["Authorization"] == "Bearer still-good"
|