Skip to content

Commit d80c25c

Browse files
committed
tests: Simplify rekorv2 tests
* Don't pretend these are unit tests: just have something simple that proves the client roughly does what it should * We should have real unit tests (that don't require the whole staging infra as current tests do) and integration tests but this is what I have now Signed-off-by: Jussi Kukkonen <[email protected]>
1 parent 9b60fb6 commit d80c25c

File tree

1 file changed

+25
-171
lines changed

1 file changed

+25
-171
lines changed
Lines changed: 25 additions & 171 deletions
Original file line numberDiff line numberDiff line change
@@ -1,83 +1,27 @@
11
import hashlib
2-
import secrets
32

43
import pytest
54

65
from sigstore import dsse
7-
from sigstore._internal.rekor.client import DEFAULT_REKOR_URL, STAGING_REKOR_URL
86
from sigstore._internal.rekor.client_v2 import (
9-
DEFAULT_KEY_DETAILS,
10-
Certificate,
11-
Hashed,
127
LogEntry,
138
RekorV2Client,
14-
common_v1,
15-
serialization,
16-
v2,
17-
v2_intoto,
189
)
19-
from sigstore._utils import sha256_digest
2010
from sigstore.models import rekor_v1
21-
from sigstore.sign import ec
2211

2312
ALPHA_REKOR_V2_URL = "https://log2025-alpha1.rekor.sigstage.dev"
24-
LOCAL_REKOR_V2_URL = "http://localhost:3000"
2513

26-
27-
# TODO: add staging and production URLs when available,
28-
# and local after using scaffolding/setup-sigstore-env action
29-
@pytest.fixture(
30-
scope="session",
31-
params=[
32-
ALPHA_REKOR_V2_URL,
33-
pytest.param(STAGING_REKOR_URL, marks=pytest.mark.xfail),
34-
pytest.param(DEFAULT_REKOR_URL, marks=pytest.mark.skip),
35-
pytest.param(LOCAL_REKOR_V2_URL, marks=pytest.mark.skip),
36-
],
37-
)
38-
def client(request) -> RekorV2Client:
39-
"""
40-
Returns a RekorV2Client. This fixture is paramaterized to return clients with various URLs.
41-
Test fuctions that consume this fixture will run once for each URL.
42-
"""
43-
return RekorV2Client(base_url=request.param)
44-
45-
46-
@pytest.fixture(scope="session")
47-
def sample_cert_and_private_key(
48-
staging,
49-
) -> tuple[Certificate, ec.EllipticCurvePrivateKey]:
50-
"""
51-
Returns a sample Certificate and ec.EllipticCurvePrivateKey.
52-
"""
14+
@pytest.mark.staging
15+
@pytest.mark.ambient_oidc
16+
def test_rekor_v2_create_entry_dsse(staging):
17+
# This is not a real unit test: it requires not only staging rekor but also TUF
18+
# fulcio and oidc -- maybe useful only until we have real integration tests in place
5319
sign_ctx_cls, _, identity = staging
54-
with sign_ctx_cls().signer(identity) as signer:
55-
return signer._signing_cert(), signer._private_key
5620

21+
# Hack to run Signer.sign() with staging rekor v2
22+
sign_ctx = sign_ctx_cls()
23+
sign_ctx._rekor = RekorV2Client(ALPHA_REKOR_V2_URL)
5724

58-
@pytest.fixture(scope="session")
59-
def sample_hashed_rekord_request_materials(
60-
sample_cert_and_private_key,
61-
) -> tuple[Hashed, bytes, Certificate]:
62-
"""
63-
Creates materials needed for `RekorV2Client._build_hashed_rekord_create_entry_request`.
64-
"""
65-
cert, private_key = sample_cert_and_private_key
66-
hashed_input = sha256_digest(secrets.token_bytes(32))
67-
signature = private_key.sign(
68-
hashed_input.digest, ec.ECDSA(hashed_input._as_prehashed())
69-
)
70-
return hashed_input, signature, cert
71-
72-
73-
@pytest.fixture(scope="session")
74-
def sample_dsse_request_materials(
75-
sample_cert_and_private_key,
76-
) -> tuple[dsse.Envelope, Certificate]:
77-
"""
78-
Creates materials needed for `RekorV2Client._build_dsse_create_entry_request`.
79-
"""
80-
cert, private_key = sample_cert_and_private_key
8125
stmt = (
8226
dsse.StatementBuilder()
8327
.subjects(
@@ -95,119 +39,29 @@ def sample_dsse_request_materials(
9539
}
9640
)
9741
).build()
98-
envelope = dsse._sign(key=private_key, stmt=stmt)
99-
return envelope, cert
100-
101-
102-
@pytest.fixture(scope="session")
103-
def sample_hashed_rekord_create_entry_request(
104-
sample_hashed_rekord_request_materials,
105-
) -> v2.CreateEntryRequest:
106-
"""
107-
Returns a sample `CreateEntryRequest` for for hashedrekor.
108-
"""
109-
hashed_input, signature, cert = sample_hashed_rekord_request_materials
110-
return RekorV2Client._build_hashed_rekord_request(
111-
hashed_input=hashed_input,
112-
signature=signature,
113-
certificate=cert,
114-
)
11542

43+
with sign_ctx.signer(identity) as signer:
44+
bundle = signer.sign_dsse(stmt)
11645

117-
@pytest.fixture(scope="session")
118-
def sample_dsse_create_entry_request(
119-
sample_dsse_request_materials,
120-
) -> v2.CreateEntryRequest:
121-
"""
122-
Returns a sample `CreateEntryRequest` for for dsse.
123-
"""
124-
envelope, cert = sample_dsse_request_materials
125-
return RekorV2Client._build_dsse_request(envelope=envelope, certificate=cert)
46+
assert isinstance(bundle.log_entry, LogEntry)
47+
assert isinstance(bundle.log_entry._to_rekor(), rekor_v1.TransparencyLogEntry)
12648

12749

50+
@pytest.mark.staging
12851
@pytest.mark.ambient_oidc
129-
def test_build_hashed_rekord_create_entry_request(
130-
sample_hashed_rekord_request_materials,
131-
):
132-
"""
133-
Ensures that we produce the request `CreateEntryRequest` correctly for hashedrekords.
134-
"""
135-
hashed_input, signature, cert = sample_hashed_rekord_request_materials
136-
expected_request = v2.CreateEntryRequest(
137-
hashed_rekord_request_v0_0_2=v2.HashedRekordRequestV002(
138-
digest=hashed_input.digest,
139-
signature=v2.Signature(
140-
content=signature,
141-
verifier=v2.Verifier(
142-
x509_certificate=common_v1.X509Certificate(
143-
raw_bytes=cert.public_bytes(encoding=serialization.Encoding.DER)
144-
),
145-
key_details=DEFAULT_KEY_DETAILS,
146-
),
147-
),
148-
)
149-
)
150-
actual_request = RekorV2Client._build_hashed_rekord_request(
151-
hashed_input=hashed_input,
152-
signature=signature,
153-
certificate=cert,
154-
)
155-
assert expected_request == actual_request
52+
def test_rekor_v2_create_entry_hashed_rekord(staging):
53+
# This is not a real unit test: it requires not only staging rekor but also TUF
54+
# fulcio and oidc -- maybe useful only until we have real integration tests in place
55+
sign_ctx_cls, _, identity = staging
15656

57+
# Hack to run Signer.sign() with staging rekor v2
58+
sign_ctx = sign_ctx_cls()
59+
sign_ctx._rekor = RekorV2Client(ALPHA_REKOR_V2_URL)
15760

158-
@pytest.mark.ambient_oidc
159-
def test_build_dsse_create_entry_request(sample_dsse_request_materials):
160-
"""
161-
Ensures that we produce the request `CreateEntryRequest` correctly for dsses.
162-
"""
163-
envelope, cert = sample_dsse_request_materials
164-
expected_request = v2.CreateEntryRequest(
165-
dsse_request_v0_0_2=v2.DsseRequestV002(
166-
envelope=v2_intoto.Envelope(
167-
payload=envelope._inner.payload,
168-
payload_type=envelope._inner.payload_type,
169-
signatures=[
170-
v2_intoto.Signature(
171-
keyid=signature.keyid,
172-
sig=signature.sig,
173-
)
174-
for signature in envelope._inner.signatures
175-
],
176-
),
177-
verifiers=[
178-
v2.Verifier(
179-
x509_certificate=common_v1.X509Certificate(
180-
raw_bytes=cert.public_bytes(encoding=serialization.Encoding.DER)
181-
),
182-
key_details=DEFAULT_KEY_DETAILS,
183-
)
184-
],
185-
)
186-
)
187-
actual_request = RekorV2Client._build_dsse_request(
188-
envelope=envelope, certificate=cert
189-
)
190-
assert expected_request == actual_request
19161

62+
with sign_ctx.signer(identity) as signer:
63+
bundle = signer.sign_artifact(b"")
64+
65+
assert isinstance(bundle.log_entry, LogEntry)
66+
assert isinstance(bundle.log_entry._to_rekor(), rekor_v1.TransparencyLogEntry)
19267

193-
@pytest.mark.parametrize(
194-
"sample_create_entry_request",
195-
[
196-
sample_hashed_rekord_create_entry_request.__name__,
197-
sample_dsse_create_entry_request.__name__,
198-
],
199-
)
200-
@pytest.mark.ambient_oidc
201-
def test_create_entry(
202-
request: pytest.FixtureRequest,
203-
sample_create_entry_request: str,
204-
client: RekorV2Client,
205-
):
206-
"""
207-
Sends a request to RekorV2 and ensure's the response is parseable to a `LogEntry` and a `TransparencyLogEntry`.
208-
"""
209-
log_entry = client.create_entry(
210-
request.getfixturevalue(sample_create_entry_request)
211-
)
212-
assert isinstance(log_entry, LogEntry)
213-
assert isinstance(log_entry._to_rekor(), rekor_v1.TransparencyLogEntry)

0 commit comments

Comments
 (0)