Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ dev = [
"mypy>=1.1,<3.0",
# NOTE(ww): ruff is under active development, so we pin conservatively here
# and let Dependabot periodically perform this update.
"ruff<0.15.23",
"ruff<0.16.1",
"types-pyOpenSSL",
"mkdocs-material[imaging]",
"mkdocstrings-python",
Expand Down Expand Up @@ -130,7 +130,14 @@ exclude_dirs = ["./test"]
[tool.ruff.lint]
extend-select = ["I", "UP"]
ignore = [
"TRY004", # invalid type does not always lead to TypeError in this code base
"UP007", # https://github.com/pydantic/pydantic/issues/4146
"UP011",
"UP015",
]

[tool.ruff.lint.per-file-ignores]
"test/**" = [
"BLE001", # blind exception handling is fine in tests
"S110", # try-except-pass is fine in tests
]
25 changes: 14 additions & 11 deletions sigstore/_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -718,13 +718,13 @@ def _sign_file_threaded(
predicate=predicate,
)
result = signer.sign_dsse(statement_builder.build())
except ExpiredIdentity as exp_identity:
except ExpiredIdentity:
_logger.error("Signature failed: identity token has expired")
raise exp_identity
raise

except ExpiredCertificate as exp_certificate:
except ExpiredCertificate:
_logger.error("Signature failed: Fulcio signing certificate has expired")
raise exp_certificate
raise

_logger.info(
f"Transparency log entry created at index: {result.log_entry._inner.log_index}"
Expand Down Expand Up @@ -800,7 +800,7 @@ def _sign_common(
for job in futures.as_completed(jobs):
job.result()

for file, outputs in output_map.items():
for outputs in output_map.values():
if outputs.signature is not None:
print(f"Signature written to {outputs.signature}")
if outputs.certificate is not None:
Expand Down Expand Up @@ -973,12 +973,15 @@ def _collect_verification_state(
)

# Fail if digest input is not used with `--bundle` or both `--certificate` and `--signature`.
if any(isinstance(x, Hashed) for x in args.files_or_digest):
if not args.bundle and not (args.certificate and args.signature):
_invalid_arguments(
args,
"verifying a digest input (sha256:*) needs either --bundle or both --certificate and --signature",
)
if (
any(isinstance(x, Hashed) for x in args.files_or_digest)
and not args.bundle
and not (args.certificate and args.signature)
):
_invalid_arguments(
args,
"verifying a digest input (sha256:*) needs either --bundle or both --certificate and --signature",
)

# Fail if `--certificate` or `--signature` is used with `--offline`.
if args.offline and (args.certificate or args.signature):
Expand Down
2 changes: 0 additions & 2 deletions sigstore/_internal/fulcio/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,8 +67,6 @@ class FulcioClientError(Exception):
Raised on any error in the Fulcio client.
"""

pass


class _Endpoint(ABC):
def __init__(self, url: str, session: requests.Session) -> None:
Expand Down
2 changes: 1 addition & 1 deletion sigstore/_internal/merkle.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ def _decomp_inclusion_proof(index: int, size: int) -> tuple[int, int]:
"""

inner = (index ^ (size - 1)).bit_length()
border = bin(index >> inner).count("1")
border = (index >> inner).bit_count()
return inner, border


Expand Down
2 changes: 1 addition & 1 deletion sigstore/_internal/oidc/oauth.py
Original file line number Diff line number Diff line change
Expand Up @@ -156,7 +156,7 @@ def do_GET(self) -> None:
_logger.debug(f"{self.path} unavailable (teardown)")
self.send_response(404)
self.end_headers()
return None
return

r = urllib.parse.urlsplit(self.path)

Expand Down
5 changes: 1 addition & 4 deletions sigstore/_internal/rekor/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ def __init__(self, http_error: requests.HTTPError):
try:
error = rekor_types.Error.model_validate_json(http_error.response.text)
super().__init__(f"{error.code}: {error.message}")
except Exception:
except Exception: # noqa: BLE001
super().__init__(
f"Rekor returned an unknown error with HTTP {http_error.response.status_code}"
)
Expand All @@ -76,7 +76,6 @@ def create_entry(
"""
Submit the request to Rekor.
"""
pass

@classmethod
@abstractmethod
Expand All @@ -86,7 +85,6 @@ def _build_hashed_rekord_request(
"""
Construct a hashed rekord request to submit to Rekor.
"""
pass

@classmethod
@abstractmethod
Expand All @@ -96,7 +94,6 @@ def _build_dsse_request(
"""
Construct a dsse request to submit to Rekor.
"""
pass


# TODO: This should probably live somewhere better.
Expand Down
2 changes: 0 additions & 2 deletions sigstore/_internal/timestamp.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,8 +58,6 @@ class TimestampError(Exception):
A generic error in the TimestampAuthority client.
"""

pass


class TimestampAuthorityClient:
"""Internal client to deal with a Timestamp Authority"""
Expand Down
7 changes: 4 additions & 3 deletions sigstore/_internal/trust.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,9 @@
from pathlib import Path
from typing import ClassVar, NewType

import cryptography.hazmat.primitives.asymmetric.padding as padding
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec, ed25519, rsa
from cryptography.hazmat.primitives.asymmetric import ec, ed25519, padding, rsa
from cryptography.x509 import (
Certificate,
load_der_x509_certificate,
Expand Down Expand Up @@ -147,10 +146,12 @@ class Keyring:
Represents a set of keys, each of which is a potentially valid verifier.
"""

def __init__(self, public_keys: list[common_v1.PublicKey] = []):
def __init__(self, public_keys: list[common_v1.PublicKey] | None = None):
"""
Create a new `Keyring`, with `keys` as the initial set of verifying keys.
"""
if public_keys is None:
public_keys = []
self._keyring: dict[KeyID, Key] = {}

for public_key in public_keys:
Expand Down
2 changes: 0 additions & 2 deletions sigstore/dsse/_predicate.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,6 @@ class Predicate(BaseModel):
Base model for in-toto predicates
"""

pass


class _SLSAConfigBase(BaseModel):
"""
Expand Down
9 changes: 3 additions & 6 deletions sigstore/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@
)
from sigstore._internal.tuf import DEFAULT_TUF_URL, STAGING_TUF_URL, TrustUpdater
from sigstore._utils import KeyID, cert_is_leaf, cert_is_root_ca, is_timerange_valid
from sigstore.errors import Error, MetadataError, TUFError, VerificationError
from sigstore.errors import Error, MetadataError, VerificationError

# Versions supported by this client
REKOR_VERSIONS = [1, 2]
Expand Down Expand Up @@ -965,11 +965,8 @@ def from_tuf(
tr_path = updater.get_trusted_root_path()
inner_tr = trustroot_v1.TrustedRoot.from_json(Path(tr_path).read_bytes())

try:
sc_path = updater.get_signing_config_path()
inner_sc = trustroot_v1.SigningConfig.from_json(Path(sc_path).read_bytes())
except TUFError as e:
raise e
sc_path = updater.get_signing_config_path()
inner_sc = trustroot_v1.SigningConfig.from_json(Path(sc_path).read_bytes())

return cls(
trustroot_v1.ClientTrustConfig(
Expand Down
6 changes: 3 additions & 3 deletions sigstore/oidc.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@
from sigstore._internal import USER_AGENT
from sigstore.errors import Error, NetworkError

_logger = logging.getLogger(__name__)

# See: https://github.com/sigstore/fulcio/blob/b2186c0/pkg/config/config.go#L182-L201
_KNOWN_OIDC_ISSUERS = {
"https://accounts.google.com": "email",
Expand Down Expand Up @@ -228,8 +230,6 @@ class IssuerError(Exception):
Raised on any communication or format error with an OIDC issuer.
"""

pass


class Issuer:
"""
Expand Down Expand Up @@ -340,7 +340,7 @@ def identity_token( # nosec: B107
client_id,
client_secret,
)
logging.debug(f"PAYLOAD: data={data}")
_logger.debug(f"PAYLOAD: data={data}")
try:
resp = self.session.post(
self.oidc_config.token_endpoint,
Expand Down
2 changes: 1 addition & 1 deletion sigstore/sign.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@
from contextlib import contextmanager
from datetime import datetime, timezone

import cryptography.x509 as x509
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from sigstore_models.common.v1 import HashOutput, MessageSignature
Expand Down
2 changes: 1 addition & 1 deletion test/assets/x509/build-testcases.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ def _keypair(priv_key_file: Path):
_ROOT_PUBKEY, _ROOT_PRIVKEY = _keypair(_HERE / "root-privkey.pem")
_NONROOT_PUBKEY, _ = _keypair(_HERE / "nonroot-privkey.pem")

_NOT_VALID_BEFORE_DATE = datetime.datetime(2023, 1, 1)
_NOT_VALID_BEFORE_DATE = datetime.datetime(2023, 1, 1, tzinfo=datetime.timezone.utc)
_A_VERY_LONG_TIME = datetime.timedelta(days=365 * 1000)


Expand Down
2 changes: 1 addition & 1 deletion test/unit/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ def target_path(self, name: str) -> Path:
try:
path = next(matches)
except StopIteration as e:
raise Exception(f"Unable to match {name} in targets/") from e
raise RuntimeError(f"Unable to match {name} in targets/") from e

if next(matches, None) is None:
return path
Expand Down
2 changes: 1 addition & 1 deletion test/unit/internal/test_trust.py
Original file line number Diff line number Diff line change
Expand Up @@ -244,7 +244,7 @@ def test_bad_media_type(self, asset):

def test_trust_root_tuf_offline(mock_staging_tuf, tuf_dirs):
# start with empty target cache, empty local metadata dir
data_dir, cache_dir = tuf_dirs
data_dir, _cache_dir = tuf_dirs

# keep track of requests the TrustUpdater invoked by TrustedRoot makes
reqs, fail_reqs = mock_staging_tuf
Expand Down
43 changes: 19 additions & 24 deletions test/unit/test_oidc.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,15 +19,19 @@
from sigstore import oidc


@pytest.fixture
def now() -> int:
return int(datetime.datetime.now(tz=datetime.timezone.utc).timestamp())


class TestIdentityToken:
def test_invalid_jwt(self):
with pytest.raises(
oidc.IdentityError, match="Identity token is malformed or missing claims"
):
oidc.IdentityToken("invalid jwt")

def test_missing_iss(self, dummy_jwt):
now = int(datetime.datetime.now().timestamp())
def test_missing_iss(self, dummy_jwt, now: int):
jwt = dummy_jwt(
{
"aud": "sigstore",
Expand All @@ -43,8 +47,7 @@ def test_missing_iss(self, dummy_jwt):
):
oidc.IdentityToken(jwt)

def test_missing_aud(self, dummy_jwt):
now = int(datetime.datetime.now().timestamp())
def test_missing_aud(self, dummy_jwt, now: int):
jwt = dummy_jwt(
{
"sub": "fakesubject",
Expand All @@ -61,8 +64,7 @@ def test_missing_aud(self, dummy_jwt):
oidc.IdentityToken(jwt)

@pytest.mark.parametrize("aud", (None, "not-sigstore"))
def test_invalid_aud(self, dummy_jwt, aud):
now = int(datetime.datetime.now().timestamp())
def test_invalid_aud(self, dummy_jwt, aud, now: int):
jwt = dummy_jwt(
{
"aud": aud,
Expand All @@ -79,8 +81,7 @@ def test_invalid_aud(self, dummy_jwt, aud):
):
oidc.IdentityToken(jwt)

def test_missing_iat(self, dummy_jwt):
now = int(datetime.datetime.now().timestamp())
def test_missing_iat(self, dummy_jwt, now: int):
jwt = dummy_jwt(
{
"aud": "sigstore",
Expand All @@ -97,8 +98,7 @@ def test_missing_iat(self, dummy_jwt):
oidc.IdentityToken(jwt)

@pytest.mark.parametrize("iat", (None, "not-an-int"))
def test_invalid_iat(self, dummy_jwt, iat):
now = int(datetime.datetime.now().timestamp())
def test_invalid_iat(self, dummy_jwt, iat, now: int):
jwt = dummy_jwt(
{
"aud": "sigstore",
Expand All @@ -115,8 +115,7 @@ def test_invalid_iat(self, dummy_jwt, iat):
):
oidc.IdentityToken(jwt)

def test_missing_nbf_ok(self, dummy_jwt):
now = int(datetime.datetime.now().timestamp())
def test_missing_nbf_ok(self, dummy_jwt, now: int):
jwt = dummy_jwt(
{
"aud": "sigstore",
Expand All @@ -129,8 +128,7 @@ def test_missing_nbf_ok(self, dummy_jwt):

assert oidc.IdentityToken(jwt) is not None

def test_invalid_nbf(self, dummy_jwt):
now = int(datetime.datetime.now().timestamp())
def test_invalid_nbf(self, dummy_jwt, now: int):
jwt = dummy_jwt(
{
"aud": "sigstore",
Expand All @@ -148,8 +146,7 @@ def test_invalid_nbf(self, dummy_jwt):
):
oidc.IdentityToken(jwt)

def test_missing_exp(self, dummy_jwt):
now = int(datetime.datetime.now().timestamp())
def test_missing_exp(self, dummy_jwt, now: int):
jwt = dummy_jwt(
{
"aud": "sigstore",
Expand All @@ -165,8 +162,7 @@ def test_missing_exp(self, dummy_jwt):
):
oidc.IdentityToken(jwt)

def test_invalid_exp(self, dummy_jwt):
now = int(datetime.datetime.now().timestamp())
def test_invalid_exp(self, dummy_jwt, now: int):
jwt = dummy_jwt(
{
"aud": "sigstore",
Expand All @@ -187,8 +183,7 @@ def test_invalid_exp(self, dummy_jwt):
@pytest.mark.parametrize(
"iss", [k for k, v in oidc._KNOWN_OIDC_ISSUERS.items() if v != "sub"]
)
def test_missing_identity_claim(self, dummy_jwt, iss):
now = int(datetime.datetime.now().timestamp())
def test_missing_identity_claim(self, dummy_jwt, iss, now: int):
jwt = dummy_jwt(
{
"aud": "sigstore",
Expand All @@ -207,8 +202,7 @@ def test_missing_identity_claim(self, dummy_jwt, iss):
oidc.IdentityToken(jwt)

@pytest.mark.parametrize("fed", ("notadict", {"connector_id": 123}))
def test_invalid_federated_claims(self, dummy_jwt, fed):
now = int(datetime.datetime.now().timestamp())
def test_invalid_federated_claims(self, dummy_jwt, fed, now: int):
jwt = dummy_jwt(
{
"aud": "sigstore",
Expand Down Expand Up @@ -248,8 +242,9 @@ def test_invalid_federated_claims(self, dummy_jwt, fed):
("hxxps://unknown.issuer.example.com/auth", "sub", "some-subject", None),
],
)
def test_ok(self, dummy_jwt, iss, identity_claim, identity_value, fed_iss):
now = int(datetime.datetime.now().timestamp())
def test_ok(
self, dummy_jwt, iss, identity_claim, identity_value, fed_iss, now: int
):
jwt = dummy_jwt(
{
"aud": "sigstore",
Expand Down
Loading
Loading