From 74cccdbfc780556a75f4d44da43302613b848be4 Mon Sep 17 00:00:00 2001 From: Victor Shyba Date: Sun, 30 Dec 2018 21:23:03 -0300 Subject: [PATCH] simplify detached signing and verifying --- lbrynet/schema/claim.py | 12 +++--------- lbrynet/schema/signer.py | 39 +++++++++++++++---------------------- lbrynet/schema/validator.py | 32 +++++++++--------------------- 3 files changed, 28 insertions(+), 55 deletions(-) diff --git a/lbrynet/schema/claim.py b/lbrynet/schema/claim.py index 6c392c539..aebcd0449 100644 --- a/lbrynet/schema/claim.py +++ b/lbrynet/schema/claim.py @@ -166,21 +166,15 @@ class ClaimDict(OrderedDict): def sign(self, private_key, claim_address, cert_claim_id, curve=NIST256p, name=None): signer = get_signer(curve).load_pem(private_key) - if name: - signature = signer.detached_sign_stream_claim(self, claim_address, cert_claim_id, name) - return ClaimDict(self, detached_signature=signature) - signed = signer.sign_stream_claim(self, claim_address, cert_claim_id) - return ClaimDict.load_protobuf(signed) + signed, signature = signer.sign_stream_claim(self, claim_address, cert_claim_id, name) + return ClaimDict.load_protobuf(signed, signature) def validate_signature(self, claim_address, certificate, name=None): if isinstance(certificate, ClaimDict): certificate = certificate.protobuf curve = CURVE_NAMES[certificate.certificate.keyType] validator = get_validator(curve).load_from_certificate(certificate, self.certificate_id) - if self.detached_signature: - assert name is not None, "Name is required for verifying detached signatures." - return validator.validate_detached_claim_signature(self, claim_address, name) - return validator.validate_claim_signature(self, claim_address) + return validator.validate_claim_signature(self, claim_address, name) def validate_private_key(self, private_key, certificate_id): certificate = self.protobuf diff --git a/lbrynet/schema/signer.py b/lbrynet/schema/signer.py index 80d3089fe..e61d8efdb 100644 --- a/lbrynet/schema/signer.py +++ b/lbrynet/schema/signer.py @@ -16,6 +16,7 @@ class NIST_ECDSASigner(object): CURVE_NAME = None HASHFUNC = hashlib.sha256 HASHFUNC_NAME = SHA256 + DETACHED = False def __init__(self, private_key): self._private_key = private_key @@ -46,16 +47,26 @@ class NIST_ECDSASigner(object): def generate(cls): return cls(ecdsa.SigningKey.generate(curve=cls.CURVE, hashfunc=cls.HASHFUNC_NAME)) - def sign_stream_claim(self, claim, claim_address, cert_claim_id): + def sign_stream_claim(self, claim, claim_address, cert_claim_id, name): + to_sign = bytearray() + if self.DETACHED: + assert name, "Name is required for detached signatures" + assert self.CURVE_NAME == SECP256k1, f"Only SECP256k1 is supported, not: {self.CURVE_NAME}" + to_sign.extend(name.lower().encode()) + validate_claim_id(cert_claim_id) + raw_cert_id = binascii.unhexlify(cert_claim_id) decoded_addr = decode_address(claim_address) - to_sign = bytearray() to_sign.extend(decoded_addr) to_sign.extend(claim.serialized_no_signature) - to_sign.extend(binascii.unhexlify(cert_claim_id)) + to_sign.extend(raw_cert_id) digest = self.HASHFUNC(to_sign).digest() + if self.DETACHED: + return claim.protobuf_dict, Signature( + self.private_key.sign_digest_deterministic(digest, hashfunc=self.HASHFUNC), raw_cert_id + ) if not isinstance(self.private_key, ecdsa.SigningKey): raise Exception("Not given a signing key") @@ -63,7 +74,7 @@ class NIST_ECDSASigner(object): "version": V_0_0_1, "signatureType": self.CURVE_NAME, "signature": self.private_key.sign_digest_deterministic(digest, hashfunc=self.HASHFUNC), - "certificateId": binascii.unhexlify(cert_claim_id) + "certificateId": raw_cert_id } msg = { @@ -72,25 +83,7 @@ class NIST_ECDSASigner(object): "publisherSignature": sig_dict } - return Claim.load(msg) - - def detached_sign_stream_claim(self, claim, claim_address, cert_claim_id, name: str): - assert self.CURVE_NAME == SECP256k1, f"Only SECP256k1 is supported, not: {self.CURVE_NAME}" - validate_claim_id(cert_claim_id) - if not isinstance(self.private_key, ecdsa.SigningKey): - raise Exception("Not given a signing key") - decoded_addr = decode_address(claim_address) - name = name.lower().encode() - raw_cert_id = binascii.unhexlify(cert_claim_id) - - to_sign = bytearray() - to_sign.extend(name) - to_sign.extend(decoded_addr) - to_sign.extend(claim.serialized_no_signature) - to_sign.extend(raw_cert_id) - - digest = self.HASHFUNC(to_sign).digest() - return Signature(self.private_key.sign_digest_deterministic(digest, hashfunc=self.HASHFUNC), raw_cert_id) + return Claim.load(msg), None class NIST256pSigner(NIST_ECDSASigner): diff --git a/lbrynet/schema/validator.py b/lbrynet/schema/validator.py index d344ecc80..e366d092a 100644 --- a/lbrynet/schema/validator.py +++ b/lbrynet/schema/validator.py @@ -76,35 +76,21 @@ class Validator: from ecdsa import BadSignatureError raise BadSignatureError - def validate_detached_claim_signature(self, claim, claim_address, name): + def validate_claim_signature(self, claim, claim_address, name): + to_sign = bytearray() + if claim.detached_signature: + assert name is not None, "Name is required for verifying detached signatures." + to_sign.extend(name.lower().encode()) + signature = claim.detached_signature.raw_signature + else: + # extract and serialize the stream from the claim, then check the signature + signature = binascii.unhexlify(claim.signature) decoded_address = decode_address(claim_address) - # extract and serialize the stream from the claim, then check the signature - signature = claim.detached_signature.raw_signature if signature is None: raise Exception("No signature to validate") - name = name.lower().encode() - - to_sign = bytearray() - to_sign.extend(name) - to_sign.extend(decoded_address) - to_sign.extend(claim.serialized_no_signature) - to_sign.extend(binascii.unhexlify(self.certificate_claim_id)) - - return self.validate_signature(self.HASHFUNC(to_sign).digest(), signature) - - def validate_claim_signature(self, claim, claim_address): - decoded_address = decode_address(claim_address) - - # extract and serialize the stream from the claim, then check the signature - signature = binascii.unhexlify(claim.signature) - - if signature is None: - raise Exception("No signature to validate") - - to_sign = bytearray() to_sign.extend(decoded_address) to_sign.extend(claim.serialized_no_signature) to_sign.extend(binascii.unhexlify(self.certificate_claim_id))