Apply new ruff rules (sigstore#1319)

Apply ruff rules

F822 Undefined name in `__all__`
EXE001 Shebang is present but file is not executable
PYI019 Use `Self` instead of custom TypeVar
PYI030 Multiple literal members in a union
PYI036 The first argument in `__exit__` should be annotated with `object` or `type[BaseException] | None`
PYI036 The second argument in `__exit__` should be annotated with `object` or `BaseException | None`
PYI036 The third argument in `__exit__` should be annotated with `object` or `types.TracebackType | None`
SIM102 Use a single `if` statement instead of nested `if` statements
SIM103 Return the negated condition directly
SIM108 Use binary operator instead of `if`-`else`-block
PERF401 Use a list comprehension to create a transformed list
UP006 Use `dict` instead of `Dict` for type annotation
UP006 Use `list` instead of `List` for type annotation
UP007 Use `X | Y` for type annotations
UP012 Unnecessary call to `encode` as UTF-8
UP031 Use format specifiers instead of percent format
UP034 Avoid extraneous parentheses
UP035 Import from `collections.abc` instead
FURB110 Replace ternary `if` expression with `or` operator
RUF010 Use explicit conversion flag
RUF036 `None` not at the end of the type annotation.
RUF100 Unused `noqa` directive

Signed-off-by: Dimitri Papadopoulos <[email protected]>

Author: DimitriPapadopoulos

docs/scripts/gen_ref_pages.py

@@ -38,10 +38,9 @@ def main(args: argparse.Namespace) -> None:
         if any(part.startswith("_") for part in module_path.parts):
             continue
 
-        if args.check:
-            if not full_doc_path.is_file():
-                print(f"File {full_doc_path} does not exist.", file=sys.stderr)
-                sys.exit(1)
+        if args.check and not full_doc_path.is_file():
+            print(f"File {full_doc_path} does not exist.", file=sys.stderr)
+            sys.exit(1)
 
         full_doc_path.parent.mkdir(parents=True, exist_ok=True)
         with full_doc_path.open("w") as f:

sigstore/_cli.py

@@ -22,7 +22,7 @@
 import sys
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Dict, NoReturn, Optional, TextIO, Union
+from typing import Any, NoReturn, Optional, TextIO, Union
 
 from cryptography.hazmat.primitives.serialization import Encoding
 from cryptography.x509 import load_pem_x509_certificate
@@ -98,7 +98,7 @@ class VerificationBundledMaterials:
 ]
 
 # Map of inputs -> outputs for signing operations
-OutputMap: TypeAlias = Dict[Path, SigningOutputs]
+OutputMap: TypeAlias = dict[Path, SigningOutputs]
 
 
 def _fatal(message: str) -> NoReturn:
@@ -200,7 +200,7 @@ def _add_shared_verification_options(group: argparse._ArgumentGroup) -> None:
 
 
 def _add_shared_oidc_options(
-    group: Union[argparse._ArgumentGroup, argparse.ArgumentParser],
+    group: argparse._ArgumentGroup | argparse.ArgumentParser,
 ) -> None:
     """
     Common OIDC options, shared between `sigstore sign` and `sigstore get-identity-token`.
@@ -833,7 +833,7 @@ def _sign(args: argparse.Namespace) -> None:
             args.bundle,
         )
 
-        output_dir = args.output_directory if args.output_directory else file.parent
+        output_dir = args.output_directory or file.parent
         if output_dir.exists() and not output_dir.is_dir():
             _invalid_arguments(
                 args, f"Output directory exists and is not a directory: {output_dir}"
@@ -895,7 +895,7 @@ 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 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,
@@ -1200,10 +1200,7 @@ def _get_identity(args: argparse.Namespace) -> Optional[IdentityToken]:
 def _fix_bundle(args: argparse.Namespace) -> None:
     # NOTE: We could support `--trusted-root` here in the future,
     # for custom Rekor instances.
-    if args.staging:
-        rekor = RekorClient.staging()
-    else:
-        rekor = RekorClient.production()
+    rekor = RekorClient.staging() if args.staging else RekorClient.production()
 
     raw_bundle = RawBundle().from_json(args.bundle.read_text())
 

sigstore/_internal/fulcio/client.py

@@ -23,7 +23,6 @@
 import logging
 from abc import ABC
 from dataclasses import dataclass
-from typing import List
 from urllib.parse import urljoin
 
 import requests
@@ -55,14 +54,14 @@ class FulcioCertificateSigningResponse:
     """Certificate response"""
 
     cert: Certificate
-    chain: List[Certificate]
+    chain: list[Certificate]
 
 
 @dataclass(frozen=True)
 class FulcioTrustBundleResponse:
     """Trust bundle response, containing a list of certificate chains"""
 
-    trust_bundle: List[List[Certificate]]
+    trust_bundle: list[list[Certificate]]
 
 
 class FulcioClientError(Exception):
@@ -151,9 +150,9 @@ def get(self) -> FulcioTrustBundleResponse:
             raise FulcioClientError from http_error
 
         trust_bundle_json = resp.json()
-        chains: List[List[Certificate]] = []
+        chains: list[list[Certificate]] = []
         for certificate_chain in trust_bundle_json["chains"]:
-            chain: List[Certificate] = []
+            chain: list[Certificate] = []
             for certificate in certificate_chain["certificates"]:
                 cert: Certificate = load_pem_x509_certificate(certificate.encode())
                 chain.append(cert)

sigstore/_internal/merkle.py

@@ -27,7 +27,6 @@
 import hashlib
 import struct
 import typing
-from typing import List, Tuple
 
 from sigstore._utils import HexStr
 from sigstore.errors import VerificationError
@@ -40,7 +39,7 @@
 _NODE_HASH_PREFIX = 1
 
 
-def _decomp_inclusion_proof(index: int, size: int) -> Tuple[int, int]:
+def _decomp_inclusion_proof(index: int, size: int) -> tuple[int, int]:
     """
     Breaks down inclusion proof for a leaf at the specified |index| in a tree of the specified
     |size| into 2 components. The splitting point between them is where paths to leaves |index| and
@@ -55,7 +54,7 @@ def _decomp_inclusion_proof(index: int, size: int) -> Tuple[int, int]:
     return inner, border
 
 
-def _chain_inner(seed: bytes, hashes: List[str], log_index: int) -> bytes:
+def _chain_inner(seed: bytes, hashes: list[str], log_index: int) -> bytes:
     """
     Computes a subtree hash for a node on or below the tree's right border. Assumes |proof| hashes
     are ordered from lower levels to upper, and |seed| is the initial subtree/leaf hash on the path
@@ -71,7 +70,7 @@ def _chain_inner(seed: bytes, hashes: List[str], log_index: int) -> bytes:
     return seed
 
 
-def _chain_border_right(seed: bytes, hashes: List[str]) -> bytes:
+def _chain_border_right(seed: bytes, hashes: list[str]) -> bytes:
     """
     Chains proof hashes along tree borders. This differs from inner chaining because |proof|
     contains only left-side subtree hashes.

sigstore/_internal/oidc/oauth.py

@@ -26,7 +26,8 @@
 import threading
 import urllib.parse
 import uuid
-from typing import Any, Dict, List, Optional, cast
+from types import TracebackType
+from typing import Any, Optional, cast
 
 from id import IdentityError
 
@@ -97,7 +98,7 @@
     </script>
   </body>
 </html>
-"""  # noqa: E501
+"""
 
 
 class _OAuthFlow:
@@ -118,7 +119,12 @@ def __enter__(self) -> _OAuthRedirectServer:
 
         return self._server
 
-    def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
         self._server.shutdown()
         self._server_thread.join()
 
@@ -200,7 +206,7 @@ def auth_endpoint(self, redirect_uri: str) -> str:
         params = self._auth_params(redirect_uri)
         return f"{self._issuer.oidc_config.authorization_endpoint}?{urllib.parse.urlencode(params)}"
 
-    def _auth_params(self, redirect_uri: str) -> Dict[str, Any]:
+    def _auth_params(self, redirect_uri: str) -> dict[str, Any]:
         return {
             "response_type": "code",
             "client_id": self._client_id,
@@ -218,7 +224,7 @@ class _OAuthRedirectServer(http.server.HTTPServer):
     def __init__(self, client_id: str, client_secret: str, issuer: Issuer) -> None:
         super().__init__(("localhost", 0), _OAuthRedirectHandler)
         self.oauth_session = _OAuthSession(client_id, client_secret, issuer)
-        self.auth_response: Optional[Dict[str, List[str]]] = None
+        self.auth_response: Optional[dict[str, list[str]]] = None
         self._is_out_of_band = False
 
     @property

sigstore/_internal/rekor/checkpoint.py

@@ -23,7 +23,6 @@
 import struct
 import typing
 from dataclasses import dataclass
-from typing import List
 
 from pydantic import BaseModel, Field, StrictStr
 
@@ -65,7 +64,7 @@ class LogCheckpoint(BaseModel):
     origin: StrictStr
     log_size: int
     log_hash: StrictStr
-    other_content: List[str]
+    other_content: list[str]
 
     @classmethod
     def from_text(cls, text: str) -> LogCheckpoint:
@@ -229,5 +228,5 @@ def verify_checkpoint(rekor_keyring: RekorKeyring, entry: LogEntry) -> None:
     if checkpoint_hash != root_hash:
         raise VerificationError(
             "Inclusion proof contains invalid root hash signature: ",
-            f"expected {str(checkpoint_hash)} got {str(root_hash)}",
+            f"expected {checkpoint_hash} got {root_hash}",
         )

sigstore/_internal/rekor/client.py

@@ -22,7 +22,7 @@
 import logging
 from abc import ABC
 from dataclasses import dataclass
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 from urllib.parse import urljoin
 
 import rekor_types
@@ -50,7 +50,7 @@ class RekorLogInfo:
     raw_data: dict
 
     @classmethod
-    def from_response(cls, dict_: Dict[str, Any]) -> RekorLogInfo:
+    def from_response(cls, dict_: dict[str, Any]) -> RekorLogInfo:
         """
         Create a new `RekorLogInfo` from the given API response.
         """

sigstore/_internal/sct.py

@@ -19,7 +19,7 @@
 import logging
 import struct
 from datetime import timezone
-from typing import List, Optional
+from typing import Optional
 
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import ec, rsa
@@ -115,7 +115,7 @@ def _pack_digitally_signed(
     # Assemble a format string with the certificate length baked in and then pack the digitally
     # signed data
     # fmt: off
-    pattern = "!BBQH%dsH" % len(signed_entry)
+    pattern = f"!BBQH{len(signed_entry)}sH"
     timestamp = sct.timestamp.replace(tzinfo=timezone.utc)
     data = struct.pack(
         pattern,
@@ -141,7 +141,7 @@ def _is_preissuer(issuer: Certificate) -> bool:
     return ExtendedKeyUsageOID.CERTIFICATE_TRANSPARENCY in ext_key_usage.value
 
 
-def _get_issuer_cert(chain: List[Certificate]) -> Certificate:
+def _get_issuer_cert(chain: list[Certificate]) -> Certificate:
     issuer = chain[0]
     if _is_preissuer(issuer):
         issuer = chain[1]
@@ -184,7 +184,7 @@ def _cert_is_ca(cert: Certificate) -> bool:
 
 def verify_sct(
     cert: Certificate,
-    chain: List[Certificate],
+    chain: list[Certificate],
     ct_keyring: CTKeyring,
 ) -> None:
     """

sigstore/_internal/trust.py

@@ -18,11 +18,12 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterable
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from enum import Enum
 from pathlib import Path
-from typing import ClassVar, Iterable, List, NewType
+from typing import ClassVar, NewType
 
 import cryptography.hazmat.primitives.asymmetric.padding as padding
 from cryptography.exceptions import InvalidSignature
@@ -159,7 +160,7 @@ class Keyring:
     Represents a set of keys, each of which is a potentially valid verifier.
     """
 
-    def __init__(self, public_keys: List[_PublicKey] = []):
+    def __init__(self, public_keys: list[_PublicKey] = []):
         """
         Create a new `Keyring`, with `keys` as the initial set of verifying keys.
         """
@@ -182,10 +183,7 @@ def verify(self, *, key_id: KeyID, signature: bytes, data: bytes) -> None:
         """
 
         key = self._keyring.get(key_id)
-        if key is not None:
-            candidates = [key]
-        else:
-            candidates = list(self._keyring.values())
+        candidates = [key] if key is not None else list(self._keyring.values())
 
         # Try to verify each candidate key. In the happy case, this will
         # be exactly one candidate.

sigstore/_internal/tuf.py

@@ -114,7 +114,7 @@ def __init__(self, url: str, offline: bool = False) -> None:
         _logger.debug(f"TUF metadata: {self._metadata_dir}")
         _logger.debug(f"TUF targets cache: {self._targets_dir}")
 
-        self._updater: None | Updater = None
+        self._updater: Updater | None = None
         if offline:
             _logger.warning(
                 "TUF repository is loaded in offline mode; updates will not be performed"