Class swarmauri_signing_secp256k1.Secp256k1EnvelopeSigner.Secp256k1EnvelopeSigner

swarmauri_signing_secp256k1.Secp256k1EnvelopeSigner.Secp256k1EnvelopeSigner

Bases: ISigning

ECDSA (secp256k1) detached signer for bytes and canonicalized envelopes.

Algorithms

• ES256K (ECDSA over secp256k1 with SHA-256)
• 'ECDSA-SHA256' accepted as a synonym

Private KeyRef accepted

• {"kind":"pem_priv","data": }
• {"kind":"pem_priv_path","path": }
• {"kind":"jwk_priv","jwk": {"kty":"EC","crv":"secp256k1","x","y","d"}}
• {"kind":"cryptography_obj","obj": EllipticCurvePrivateKey}

Verification keys (opts["pubkeys"]) accepted as: - {"kind":"pem_pub","data": } - {"kind":"pem_pub_path","path": } - {"kind":"jwk_pub","jwk": {"kty":"EC","crv":"secp256k1","x","y"}} - {"kind":"cryptography_obj","obj": EllipticCurvePublicKey} - direct PEM (bytes/str) - direct JWK mapping (dict with kty='EC', crv='secp256k1', x, y)

Signature format

• DER (ASN.1) by default; include {"fmt":"DER"} in the signature map.
• JOSE raw r||s: pass opts={"format":"RAW"} to sign/verify.

supports

supports()

Source code in swarmauri_signing_secp256k1/Secp256k1EnvelopeSigner.py

def supports(self) -> Mapping[str, Iterable[str]]:
    algs = ("ES256K", "ECDSA-SHA256")
    canons = ("json", "cbor") if _CBOR_OK else ("json",)
    return {
        "algs": algs,
        "canons": canons,
        "signs": ("bytes", "digest", "envelope", "stream"),
        "verifies": ("bytes", "digest", "envelope", "stream"),
        "envelopes": ("mapping",),
        "features": ("multi", "detached_only"),
    }

sign_bytes async

sign_bytes(key, payload, *, alg=None, opts=None)

Source code in swarmauri_signing_secp256k1/Secp256k1EnvelopeSigner.py

async def sign_bytes(
    self,
    key: KeyRef,
    payload: bytes,
    *,
    alg: Optional[Alg] = None,
    opts: Optional[Mapping[str, object]] = None,
) -> Sequence[Signature]:
    _ensure_crypto()
    if alg not in (None, "ES256K", "ECDSA-SHA256"):
        raise ValueError(
            "Unsupported alg for Secp256k1EnvelopeSigner. Use 'ES256K'."
        )
    sk = _keyref_to_private(key, passphrase=(opts or {}).get("passphrase"))
    if not isinstance(sk.curve, ec.SECP256K1):
        raise TypeError("Private key curve must be secp256k1.")
    der_sig = sk.sign(payload, ec.ECDSA(hashes.SHA256()))

    fmt = (opts or {}).get("format", "DER")
    sig_bytes = der_sig
    if isinstance(fmt, str) and fmt.upper() == "RAW":
        r, s = decode_dss_signature(der_sig)
        size_bytes = (sk.curve.key_size + 7) // 8
        sig_bytes = r.to_bytes(size_bytes, "big") + s.to_bytes(size_bytes, "big")

    kid = _kid_from_public(
        sk.public_key(),
        (opts or {}).get("kid_jwk_hint") if isinstance(opts, Mapping) else None,
    )
    return [
        _Sig(
            {
                "alg": "ES256K",
                "kid": kid,
                "sig": sig_bytes,
                "fmt": "RAW" if fmt == "RAW" else "DER",
            }
        )
    ]

verify_bytes async

verify_bytes(
    payload, signatures, *, require=None, opts=None
)

Source code in swarmauri_signing_secp256k1/Secp256k1EnvelopeSigner.py

async def verify_bytes(
    self,
    payload: bytes,
    signatures: Sequence[Signature],
    *,
    require: Optional[Mapping[str, object]] = None,
    opts: Optional[Mapping[str, object]] = None,
) -> bool:
    _ensure_crypto()
    min_signers = int((require or {}).get("min_signers", 1))
    allowed_algs = set((require or {}).get("algs", ("ES256K", "ECDSA-SHA256")))
    fmt_pref = (opts or {}).get("format", "DER")

    if not (opts and "pubkeys" in opts):
        raise ValueError(
            "verify_bytes requires opts['pubkeys'] with one or more secp256k1 public keys."
        )
    pubs: list[ec.EllipticCurvePublicKey] = []
    for entry in opts["pubkeys"]:  # type: ignore[index]
        pk = _keyref_to_public(entry)
        if not isinstance(pk.curve, ec.SECP256K1):
            raise TypeError("All verification public keys must be secp256k1.")
        pubs.append(pk)

    accepted = 0
    for sig in signatures:
        alg = sig.get("alg")
        if not isinstance(alg, str) or alg not in allowed_algs:
            continue
        sig_bytes = sig.get("sig")
        if not isinstance(sig_bytes, (bytes, bytearray)):
            continue

        any_verified = False
        for pk in pubs:
            try:
                if (isinstance(fmt_pref, str) and fmt_pref.upper() == "RAW") or (
                    isinstance(sig.get("fmt"), str)
                    and sig.get("fmt").upper() == "RAW"
                ):
                    size_bytes = (pk.curve.key_size + 7) // 8
                    if len(sig_bytes) != 2 * size_bytes:
                        raise InvalidSignature(
                            "Invalid RAW ECDSA signature length for secp256k1."
                        )
                    r = int.from_bytes(sig_bytes[:size_bytes], "big")
                    s = int.from_bytes(sig_bytes[size_bytes:], "big")
                    der = encode_dss_signature(r, s)
                    pk.verify(der, payload, ec.ECDSA(hashes.SHA256()))
                else:
                    pk.verify(bytes(sig_bytes), payload, ec.ECDSA(hashes.SHA256()))
                any_verified = True
                break
            except InvalidSignature:
                continue

        if any_verified:
            accepted += 1
            if accepted >= min_signers:
                return True

    return False

canonicalize_envelope async

canonicalize_envelope(env, *, canon=None, opts=None)

Source code in swarmauri_signing_secp256k1/Secp256k1EnvelopeSigner.py

async def canonicalize_envelope(
    self,
    env: Envelope,
    *,
    canon: Optional[Canon] = None,
    opts: Optional[Mapping[str, object]] = None,
) -> bytes:
    if canon in (None, "json"):
        return _canon_json(env)  # type: ignore[arg-type]
    if canon == "cbor":
        return _canon_cbor(env)  # type: ignore[arg-type]
    raise ValueError(f"Unsupported canon: {canon}")

sign_envelope async

sign_envelope(key, env, *, alg=None, canon=None, opts=None)

Source code in swarmauri_signing_secp256k1/Secp256k1EnvelopeSigner.py

async def sign_envelope(
    self,
    key: KeyRef,
    env: Envelope,
    *,
    alg: Optional[Alg] = None,
    canon: Optional[Canon] = None,
    opts: Optional[Mapping[str, object]] = None,
) -> Sequence[Signature]:
    payload = await self.canonicalize_envelope(env, canon=canon, opts=opts)
    return await self.sign_bytes(key, payload, alg="ES256K", opts=opts)

verify_envelope async

verify_envelope(
    env, signatures, *, canon=None, require=None, opts=None
)

Source code in swarmauri_signing_secp256k1/Secp256k1EnvelopeSigner.py

async def verify_envelope(
    self,
    env: Envelope,
    signatures: Sequence[Signature],
    *,
    canon: Optional[Canon] = None,
    require: Optional[Mapping[str, object]] = None,
    opts: Optional[Mapping[str, object]] = None,
) -> bool:
    payload = await self.canonicalize_envelope(env, canon=canon, opts=opts)
    return await self.verify_bytes(payload, signatures, require=require, opts=opts)

sign_digest abstractmethod async

sign_digest(key, digest, *, alg=None, opts=None)

Produce detached signatures over a pre-computed message digest.

Source code in swarmauri_core/signing/ISigning.py

@abstractmethod
async def sign_digest(
    self,
    key: KeyRef,
    digest: bytes,
    *,
    alg: Optional[Alg] = None,
    opts: Optional[Mapping[str, object]] = None,
) -> Sequence[Signature]:
    """Produce detached signatures over a pre-computed message digest."""
    ...

sign_stream abstractmethod async

sign_stream(key, payload, *, alg=None, opts=None)

Produce detached signatures over streaming byte payloads.

Implementations MAY buffer the stream to reuse :meth:sign_bytes.

Source code in swarmauri_core/signing/ISigning.py

@abstractmethod
async def sign_stream(
    self,
    key: KeyRef,
    payload: ByteStream,
    *,
    alg: Optional[Alg] = None,
    opts: Optional[Mapping[str, object]] = None,
) -> Sequence[Signature]:
    """
    Produce detached signatures over streaming byte payloads.

    Implementations MAY buffer the stream to reuse :meth:`sign_bytes`.
    """
    ...

verify_stream abstractmethod async

verify_stream(
    payload, signatures, *, require=None, opts=None
)

Source code in swarmauri_core/signing/ISigning.py

@abstractmethod
async def verify_stream(
    self,
    payload: ByteStream,
    signatures: Sequence[Signature],
    *,
    require: Optional[Mapping[str, object]] = None,
    opts: Optional[Mapping[str, object]] = None,
) -> bool: ...

verify_digest abstractmethod async

verify_digest(
    digest, signatures, *, require=None, opts=None
)

Verify detached signatures produced over a pre-computed digest.

Source code in swarmauri_core/signing/ISigning.py

@abstractmethod
async def verify_digest(
    self,
    digest: bytes,
    signatures: Sequence[Signature],
    *,
    require: Optional[Mapping[str, object]] = None,
    opts: Optional[Mapping[str, object]] = None,
) -> bool:
    """Verify detached signatures produced over a pre-computed digest."""
    ...