298 lines
17 KiB
Python
298 lines
17 KiB
Python
"""Testele aere-cloud (Python), oglinda celor din sdk/test/sdk.test.mjs.
|
|
|
|
python3 test_aere_cloud.py unitare, cu transport fals, fara retea
|
|
AERE_KEY=ak2800... python3 test_aere_cloud.py --live + 3 probe prin productie
|
|
"""
|
|
import hashlib
|
|
import hmac as _hmac
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from aere_cloud import AereCloud, AereCloudError, verify_webhook # noqa: E402
|
|
|
|
trecute, picate = 0, 0
|
|
def proba(nume, fn):
|
|
global trecute, picate
|
|
try:
|
|
fn()
|
|
trecute += 1
|
|
print(f" OK {nume}")
|
|
except Exception as e:
|
|
picate += 1
|
|
print(f" PICA {nume}: {e!r}")
|
|
|
|
|
|
class Transport:
|
|
"""Opener fals: inregistreaza cererile si raspunde din scenariu."""
|
|
def __init__(self, raspunsuri):
|
|
self.cereri = []
|
|
self.raspunsuri = list(raspunsuri)
|
|
|
|
def __call__(self, cerere, timeout):
|
|
self.cereri.append(cerere)
|
|
status, corp = self.raspunsuri.pop(0)
|
|
return status, json.dumps(corp).encode("utf-8")
|
|
|
|
|
|
def t_constructor():
|
|
try:
|
|
AereCloud("")
|
|
raise AssertionError("nu a ridicat pe cheie goala")
|
|
except ValueError:
|
|
pass
|
|
|
|
def t_antet():
|
|
tr = Transport([(200, {"ok": True}), (200, {"ok": True})])
|
|
c = AereCloud("ak2800.0xAB.secret", opener=tr)
|
|
c.health()
|
|
assert "x-api-key" not in {k.lower() for k in tr.cereri[0].headers}, "health nu trimite cheia"
|
|
c.account()
|
|
assert tr.cereri[1].get_header("X-api-key") == "ak2800.0xAB.secret", "account trimite cheia"
|
|
|
|
def t_rpc():
|
|
tr = Transport([(200, {"jsonrpc": "2.0", "id": 1, "result": "0x10"})])
|
|
c = AereCloud("k", opener=tr)
|
|
assert c.rpc("eth_blockNumber") == "0x10"
|
|
corp = json.loads(tr.cereri[0].data.decode())
|
|
assert corp["method"] == "eth_blockNumber" and corp["jsonrpc"] == "2.0"
|
|
tr2 = Transport([(200, {"error": {"code": -32601, "message": "nope"}})])
|
|
try:
|
|
AereCloud("k", opener=tr2).rpc("qbft_x")
|
|
raise AssertionError("eroarea RPC nu s-a ridicat")
|
|
except AereCloudError as e:
|
|
assert e.status == 200
|
|
|
|
def t_eroare_http():
|
|
tr = Transport([(403, {"error": "invalid_or_expired_key"})])
|
|
try:
|
|
AereCloud("k", opener=tr).account()
|
|
raise AssertionError("403 nu s-a ridicat")
|
|
except AereCloudError as e:
|
|
assert e.status == 403 and e.body["error"] == "invalid_or_expired_key" and e.path == "/account"
|
|
|
|
def t_query():
|
|
tr = Transport([(200, {}), (200, {}), (200, {}), (200, {})])
|
|
c = AereCloud("k", opener=tr)
|
|
c.anchors(3); c.transfers("0xAB", 7); c.anchor(15000000); c.finality()
|
|
assert tr.cereri[0].full_url.endswith("/data/anchors?limit=3")
|
|
assert tr.cereri[1].full_url.endswith("/data/transfers?address=0xAB&limit=7")
|
|
assert tr.cereri[2].full_url.endswith("/data/anchors/15000000")
|
|
assert tr.cereri[3].full_url.endswith("/data/finality")
|
|
|
|
def t_verify():
|
|
tr = Transport([(200, {"verdict": "VALID", "levels": []}), (200, {"verdict": "PARTIAL", "levels": []})])
|
|
c = AereCloud("k", opener=tr)
|
|
assert c.verify({"statement": {}, "statementHash": "0x00"}, chain=28001)["verdict"] == "VALID"
|
|
assert c.verify({"statement": {}})["verdict"] == "PARTIAL"
|
|
assert tr.cereri[0].full_url.endswith("/verify?chain=28001") and tr.cereri[0].get_method() == "POST"
|
|
assert tr.cereri[1].full_url.endswith("/verify")
|
|
assert tr.cereri[0].get_header("X-api-key") == "k", "verify cere cheia"
|
|
assert json.loads(tr.cereri[0].data.decode()) == {"statement": {}, "statementHash": "0x00"}
|
|
|
|
def t_identitate():
|
|
tr = Transport([(200, {"verdict": "PARTIAL", "notJudged": 1}), (200, {"verdict": "COMPLIANT", "compliant": True})])
|
|
c = AereCloud("k", opener=tr)
|
|
cerere = {"presentation": {"kind": "aere-presentation", "v": 1}, "audience": "https://v.example", "nonce": "n-1"}
|
|
assert c.identity_verify(cerere)["verdict"] == "PARTIAL"
|
|
assert tr.cereri[0].full_url.endswith("/identity/verify") and tr.cereri[0].get_method() == "POST"
|
|
assert tr.cereri[0].get_header("X-api-key") == "k", "identity_verify cere cheia"
|
|
assert json.loads(tr.cereri[0].data.decode()) == cerere
|
|
cc = {**cerere, "policy": {"id": "p"}, "record": True}
|
|
assert c.compliance_check(cc)["verdict"] == "COMPLIANT"
|
|
assert tr.cereri[1].full_url.endswith("/compliance/check") and json.loads(tr.cereri[1].data.decode()) == cc
|
|
|
|
def t_plati():
|
|
tr = Transport([(200, {"verdict": "REFUSE", "reasons": ["above the per-payment limit"]}), (201, {"receipt": {"instruction": {}, "envelope": {}}}), (200, {"verdict": "PARTIAL"})])
|
|
c = AereCloud("k", opener=tr)
|
|
cerere = {"policy": {"v": 1, "kind": "aere-payment-risk-policy", "id": "p"}, "inputs": {"chainId": 28001, "asset": "native", "amount": "400000000000000000", "history": [], "now": 1}}
|
|
assert c.payments_risk(cerere)["verdict"] == "REFUSE", "un REFUSE e un verdict (200), nu o eroare"
|
|
assert tr.cereri[0].full_url.endswith("/payments/risk") and tr.cereri[0].get_method() == "POST"
|
|
assert tr.cereri[0].get_header("X-api-key") == "k", "payments_risk cere cheia"
|
|
assert json.loads(tr.cereri[0].data.decode()) == cerere
|
|
cc = {**cerere, "txHash": "0x" + "11" * 32, "purpose": "order 7"}
|
|
assert "receipt" in c.payments_receipt(cc)
|
|
assert tr.cereri[1].full_url.endswith("/payments/receipt") and json.loads(tr.cereri[1].data.decode()) == cc
|
|
cv = {"receipt": {"instruction": {"a": 1}, "envelope": {"b": 2}}, "policy": cerere["policy"], "inputs": cerere["inputs"]}
|
|
assert c.payments_verify(cv)["verdict"] == "PARTIAL"
|
|
assert tr.cereri[2].full_url.endswith("/payments/verify") and json.loads(tr.cereri[2].data.decode()) == cv
|
|
assert tr.cereri[2].get_header("X-api-key") == "k", "payments_verify cere cheia"
|
|
|
|
def t_pq_sponsor():
|
|
tr = Transport([(200, {"txHash": "0xab", "account": "0xcd", "alreadyDeployed": False})])
|
|
c = AereCloud("k", opener=tr)
|
|
d = c.sponsor_create_pq_account("0x09" + "00" * 896, salt=5)
|
|
corp = json.loads(tr.cereri[0].data.decode())
|
|
assert corp["salt"] == 5 and corp["falconPubKey"].startswith("0x09")
|
|
assert tr.cereri[0].full_url.endswith("/sponsor/createPqAccount")
|
|
assert d["account"] == "0xcd"
|
|
|
|
def t_webhook():
|
|
secret, corp = "sssh", b'{"type":"pq-anchor","x":1}'
|
|
semnatura = _hmac.new(secret.encode(), corp, hashlib.sha256).hexdigest()
|
|
assert verify_webhook(corp, semnatura, secret) is True
|
|
assert verify_webhook(corp, semnatura[:-1] + ("0" if semnatura[-1] != "0" else "1"), secret) is False
|
|
assert verify_webhook(corp, semnatura + "0", secret) is False
|
|
assert verify_webhook(corp + b" ", semnatura, secret) is False
|
|
|
|
proba("constructorul cere cheia", t_constructor)
|
|
proba("cheia merge in antet, health e fara cheie", t_antet)
|
|
proba("rpc impacheteaza corect si ridica eroarea RPC", t_rpc)
|
|
proba("un status de eroare devine AereCloudError cu corpul serverului", t_eroare_http)
|
|
proba("rutele de date construiesc corect query-ul", t_query)
|
|
proba("verify (Trust API, 1.6.0): ruta, ?chain, cheia si plicul neschimbat", t_verify)
|
|
proba("identity_verify si compliance_check (1.7.0): rutele, cheia si cererea neschimbata", t_identitate)
|
|
proba("payments_risk, payments_receipt si payments_verify (1.8.0): rutele, cheia si cererea neschimbata", t_plati)
|
|
proba("sponsor_create_pq_account: corp si cale corecte", t_pq_sponsor)
|
|
proba("verify_webhook: buna trece; stricata, prelungita, corp atins pica", t_webhook)
|
|
|
|
def t_proof_readiness():
|
|
# 2026-09-17: proof(hash) cere cheia; readiness() e libera si trimite cheia doar cu attest=True; corpul e exact
|
|
h = "0x" + "ab" * 32
|
|
tr = Transport([
|
|
(200, {"hash": h, "notarized": True, "finality": "post-quantum"}),
|
|
(200, {"domain": "aere.network", "score": 75}),
|
|
(200, {"domain": "aere.network", "score": 75, "attestation": {"reportHash": h}}),
|
|
(200, {"domain": "aere.network", "score": 75, "cached": True}),
|
|
])
|
|
c = AereCloud("ak2800.0xAB.secret", opener=tr)
|
|
assert c.proof(h)["finality"] == "post-quantum"
|
|
assert tr.cereri[0].get_header("X-api-key") == "ak2800.0xAB.secret", "proof cere cheia"
|
|
assert tr.cereri[0].full_url.endswith("/proof/" + h)
|
|
c.readiness("aere.network")
|
|
assert "x-api-key" not in {k.lower() for k in tr.cereri[1].headers}, "scanarea libera nu trimite cheia"
|
|
assert json.loads(tr.cereri[1].data.decode()) == {"domain": "aere.network"}
|
|
assert c.readiness("aere.network", attest=True, fresh=True)["attestation"]["reportHash"] == h
|
|
assert tr.cereri[2].get_header("X-api-key") == "ak2800.0xAB.secret", "atestarea trimite cheia"
|
|
assert json.loads(tr.cereri[2].data.decode()) == {"domain": "aere.network", "fresh": True, "attest": True}
|
|
assert c.readiness_report("aere.network")["cached"] is True
|
|
assert "x-api-key" not in {k.lower() for k in tr.cereri[3].headers}
|
|
proba("proof + readiness: ruta, cheia doar cand se cere, corpul corect", t_proof_readiness)
|
|
|
|
def t_perimetru():
|
|
# 2026-09-18: readiness_perimeter() cere cheia; corpul poarta doar ce s-a cerut; discover=True e implicitul si nu se trimite
|
|
tr = Transport([(200, {"domain": "example.com", "summary": {"measured": 3, "complete": True}}), (200, {"domain": "example.com", "summary": {"complete": True}})])
|
|
c = AereCloud("ak2800.0xAB.secret", opener=tr)
|
|
assert c.readiness_perimeter("example.com")["summary"]["measured"] == 3
|
|
assert tr.cereri[0].get_header("X-api-key") == "ak2800.0xAB.secret", "perimetrul cere cheia"
|
|
assert tr.cereri[0].full_url.endswith("/pq/readiness/perimeter")
|
|
assert json.loads(tr.cereri[0].data.decode()) == {"domain": "example.com"}
|
|
c.readiness_perimeter("example.com", hosts=["api", "mail.example.com"], discover=False, attest=True)
|
|
assert json.loads(tr.cereri[1].data.decode()) == {"domain": "example.com", "hosts": ["api", "mail.example.com"], "discover": False, "attest": True}
|
|
from aere_cloud import __version__
|
|
import re as _re, pathlib as _pl
|
|
proiect = (_pl.Path(__file__).parent / "pyproject.toml").read_text(encoding="utf-8")
|
|
assert _re.search(r'^version = "' + _re.escape(__version__) + '"', proiect, _re.M), "__version__ si pyproject.toml spun versiuni diferite (platit 2026-09-18: 1.1.0 fata de 1.2.1)"
|
|
proba("readiness_perimeter: ruta cu cheie, corpul exact; __version__ = versiunea din pyproject", t_perimetru)
|
|
|
|
def t_export_audit():
|
|
# 2026-09-18: audit_export() intoarce octetii asa cum au venit si sha256-ul LOR, calculat local; cu cheie; un 400 se arunca
|
|
import hashlib
|
|
corp = b'{"t":"2026-09-17T10:00:00.000Z","m":"GET","p":"/v1/account","s":200}\n'
|
|
class TransportBrut(Transport):
|
|
def __call__(self, cerere, timeout):
|
|
self.cereri.append(cerere); status, d = self.raspunsuri.pop(0)
|
|
return status, d
|
|
tr = TransportBrut([(200, corp), (400, b'{"error":"bad_format"}')])
|
|
c = AereCloud("ak2800.0xAB.secret", opener=tr)
|
|
e = c.audit_export(day="2026-09-17")
|
|
assert e["bytes"] == corp and e["entries"] == 1 and e["sha256"] == "0x" + hashlib.sha256(corp).hexdigest()
|
|
assert tr.cereri[0].full_url.endswith("/account/audit?format=jsonl&day=2026-09-17"), tr.cereri[0].full_url
|
|
assert tr.cereri[0].get_header("X-api-key") == "ak2800.0xAB.secret"
|
|
try:
|
|
c.audit_export(fmt="xml"); raise AssertionError("un 400 trebuia aruncat")
|
|
except AereCloudError as x:
|
|
assert x.status == 400
|
|
proba("audit_export: octetii zilei si sha256-ul lor calculat local; 400 aruncat", t_export_audit)
|
|
|
|
def t_retentie_audit():
|
|
# 2026-09-18: audit_retention() citeste cu cheia; audit_verify_kept(): exportul pastrat fata de ziua tinuta si fata de PIATRA unei zile sterse (410)
|
|
import hashlib, json as _j
|
|
corp = b'{"t":"2026-05-01T10:00:00.000Z","m":"GET","p":"/v1/account","s":200}\n'
|
|
d = "0x" + hashlib.sha256(corp).hexdigest()
|
|
piatra = _j.dumps({"error": "audit_day_expired", "expired": {"day": "2026-05-01", "digest": d, "entries": 1, "deletedAt": "2026-09-25T12:00:01.000Z"}}).encode()
|
|
class TransportBrut(Transport):
|
|
def __call__(self, cerere, timeout):
|
|
self.cereri.append(cerere); status, r = self.raspunsuri.pop(0)
|
|
return status, r
|
|
tr = TransportBrut([(200, b'{"retentionDays":90,"hold":false}'), (410, piatra), (410, piatra), (200, corp), (200, corp), (429, b'{"error":"rate_limit"}')])
|
|
c = AereCloud("ak2800.0xAB.secret", opener=tr)
|
|
assert c.audit_retention()["retentionDays"] == 90
|
|
assert tr.cereri[0].full_url.endswith("/account/audit/retention") and tr.cereri[0].get_method() == "GET" and tr.cereri[0].get_header("X-api-key") == "ak2800.0xAB.secret"
|
|
bun = c.audit_verify_kept("2026-05-01", corp)
|
|
assert (bun["state"], bun["matches"], bun["digest"], bun["deleted_at"]) == ("expired", True, d, "2026-09-25T12:00:01.000Z"), bun
|
|
stricat = bytes([corp[0] ^ 1]) + corp[1:]
|
|
assert c.audit_verify_kept("2026-05-01", stricat)["matches"] is False, "un export modificat nu are voie sa se potriveasca cu piatra"
|
|
tinut = c.audit_verify_kept("2026-05-01", corp.decode("utf-8"))
|
|
assert (tinut["state"], tinut["matches"]) == ("held", True), tinut
|
|
assert c.audit_verify_kept("2026-05-01", stricat)["matches"] is False
|
|
try:
|
|
c.audit_verify_kept("2026-05-01", corp); raise AssertionError("un 429 trebuia aruncat, nu intors ca nepotrivire")
|
|
except AereCloudError as x:
|
|
assert x.status == 429
|
|
proba("audit_retention + audit_verify_kept: piatra unei zile sterse verifica exportul pastrat; o eroare nu e o nepotrivire", t_retentie_audit)
|
|
|
|
def t_fara_redirectari():
|
|
# 1.5.1: cu transportul IMPLICIT (urllib), pe un server local real: un 302 ajunge la apelant ca AereCloudError, iar tinta din
|
|
# Location nu primeste nimic (deci nici cheia). Transportul injectat al celorlalte probe nu vede redirectarile.
|
|
import http.server, threading
|
|
vazute = []
|
|
class H(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
vazute.append((self.path, self.headers.get("x-api-key")))
|
|
if self.path.startswith("/v1/"):
|
|
self.send_response(302); self.send_header("location", "/in-alta-parte"); self.end_headers()
|
|
else:
|
|
self.send_response(200); self.send_header("content-type", "application/json"); self.end_headers(); self.wfile.write(b"{}")
|
|
def log_message(self, *a):
|
|
pass
|
|
srv = http.server.HTTPServer(("127.0.0.1", 0), H)
|
|
fir = threading.Thread(target=srv.serve_forever, daemon=True); fir.start()
|
|
try:
|
|
c = AereCloud("ak2800.0xAB.secret", base_url=f"http://127.0.0.1:{srv.server_address[1]}/v1")
|
|
try:
|
|
c.account(); raise AssertionError("302 trebuia ridicat, nu urmat")
|
|
except AereCloudError as x:
|
|
assert x.status == 302, x.status
|
|
assert not [v for v in vazute if v[0] == "/in-alta-parte"], vazute
|
|
finally:
|
|
srv.shutdown(); srv.server_close()
|
|
proba("1.5.1: o redirectare nu e urmata cu transportul implicit; tinta din Location nu primeste cheia", t_fara_redirectari)
|
|
|
|
def _live_pq(c):
|
|
try:
|
|
c.sponsor_create_pq_account("0x07", 0)
|
|
raise AssertionError("nu a refuzat cheia scurta")
|
|
except AereCloudError as e:
|
|
assert "897" in str(e.body.get("error", "")), e.body
|
|
|
|
if "--live" in sys.argv:
|
|
cheie = os.environ.get("AERE_KEY")
|
|
if not cheie:
|
|
print(" (sar LIVE: AERE_KEY lipseste)")
|
|
else:
|
|
c = AereCloud(cheie)
|
|
def t_live_health(): assert c.health()["ok"] is True
|
|
def t_live_validatori(): assert c.validators()["count"] == 10 # al zecelea (Nethermind) din 2026-09-11
|
|
def t_live_ancora(): assert c.anchors(1)["anchors"][0]["falconSeals"] >= 6
|
|
proba("LIVE: health", t_live_health)
|
|
proba("LIVE: validators = 10", t_live_validatori)
|
|
proba("LIVE: ancora cu >=6 sigilii", t_live_ancora)
|
|
proba("LIVE: sponsor PQ refuza JSON pe cheie scurta", lambda: _live_pq(c))
|
|
|
|
# 2026-09-29: Trust API prin productie - plic minimal valid -> VALID, acelasi cu statement schimbat -> INVALID
|
|
def t_live_verify():
|
|
st = {"v": 1, "kind": "aere-proof-of-data", "subject": {"name": "example.csv"}, "sha256": "0x" + "11" * 32, "createdAt": "2026-09-26T00:00:00.000Z"}
|
|
text = json.dumps(st, separators=(",", ":"), ensure_ascii=False)
|
|
plic = {"v": 1, "kind": "aere-proof-of-data-attestation", "statement": st, "statementHash": "0x" + hashlib.sha256(text.encode("utf-8")).hexdigest()}
|
|
bun = c.verify(plic)
|
|
assert bun["verdict"] == "VALID", bun
|
|
rau = c.verify(dict(plic, statement=dict(st, sha256="0x" + "22" * 32)))
|
|
assert rau["verdict"] == "INVALID", rau
|
|
assert any(l["level"] == "integrity" and l["state"] == "FAILED" for l in rau["levels"])
|
|
proba("LIVE: verify - plic valid VALID, plic atins INVALID", t_live_verify)
|
|
|
|
print(f"\n{trecute} treceri, {picate} esecuri")
|
|
sys.exit(1 if picate else 0)
|