Nucypher
/
nucypher
mirror of https://github.com/nucypher/nucypher.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Move RevocationOrder to core.py

pull/2802/head
Bogdan Opanchuk 2021-09-28 22:53:06 -07:00
parent a995685c87
commit 98c78efbca
4 changed files with 82 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
nucypher/core.py
View File

@ -774,3 +774,77 @@ class ArrangementResponse(Versioned):
def _from_bytes_current(cls, data: bytes):
signature, = signature_splitter(data)
return cls(signature=signature)
class RevocationOrder(Versioned):
"""
Represents a string used by characters to perform a revocation on a specific Ursula.
"""
@classmethod
def author(cls,
ursula_address: ChecksumAddress,
encrypted_kfrag: MessageKit,
signer: Signer) -> 'RevocationOrder':
return cls(ursula_address=ursula_address,
encrypted_kfrag=encrypted_kfrag,
signature=signer.sign(cls._signed_payload(ursula_address, encrypted_kfrag)))
def __init__(self, ursula_address: ChecksumAddress, encrypted_kfrag: MessageKit, signature: Signature):
self.ursula_address = ursula_address
self.encrypted_kfrag = encrypted_kfrag
self.signature = signature
def __repr__(self):
return bytes(self)
def __len__(self):
return len(bytes(self))
def __eq__(self, other):
return bytes(self) == bytes(other)
@staticmethod
def _signed_payload(ursula_address, encrypted_kfrag):
return to_canonical_address(ursula_address) + bytes(encrypted_kfrag)
def verify_signature(self, alice_verifying_key: PublicKey) -> bool:
"""
Verifies the revocation was from the provided pubkey.
"""
# TODO: raise an exception instead of returning `bool`?
payload = self._signed_payload(self.ursula_address, self.encrypted_kfrag)
if not self.signature.verify(payload, alice_verifying_key):
raise InvalidSignature(f"Revocation has an invalid signature: {self.signature}")
return True
@classmethod
def _brand(cls) -> bytes:
return b'Revo'
@classmethod
def _version(cls) -> Tuple[int, int]:
return 1, 0
@classmethod
def _old_version_handlers(cls) -> Dict:
return {}
def _body(self) -> bytes:
return to_canonical_address(self.ursula_checksum_address) + bytes(self.encrypted_kfrag)
def _payload(self) -> bytes:
return self._signed_payload(self.ursula_address, self.encrypted_kfrag) + bytes(self.signature)
@classmethod
def _from_bytes_current(cls, data):
splitter = BytestringSplitter(
checksum_address_splitter, # ursula canonical address
(bytes, Versioned._HEADER_SIZE+AuthorizedKeyFrag.SERIALIZED_SIZE), # MessageKit version header + versioned ekfrag
signature_splitter
)
ursula_canonical_address, ekfrag, signature = splitter(data)
ursula_address = to_checksum_address(ursula_canonical_address)
return cls(ursula_address=ursula_address,
encrypted_kfrag=ekfrag,
signature=signature)

3
nucypher/network/server.py
View File

@ -27,7 +27,7 @@ from flask import Flask, Response, jsonify, request
from mako import exceptions as mako_exceptions
from mako.template import Template
from nucypher.core import AuthorizedKeyFrag, ReencryptionRequest, Arrangement, ArrangementResponse
from nucypher.core import AuthorizedKeyFrag, ReencryptionRequest, Arrangement, ArrangementResponse, RevocationOrder
from nucypher.blockchain.eth.utils import period_to_epoch
from nucypher.config.constants import MAX_UPLOAD_CONTENT_LENGTH
@ -39,7 +39,6 @@ from nucypher.datastore.models import ReencryptionRequest as ReencryptionRequest
from nucypher.network import LEARNING_LOOP_VERSION
from nucypher.network.exceptions import NodeSeemsToBeDown
from nucypher.network.protocols import InterfaceInfo
from nucypher.policy.revocation import RevocationOrder
from nucypher.utilities.logging import Logger
HERE = BASE_DIR = Path(__file__).parent

96
nucypher/policy/revocation.py
View File

@ -16,103 +16,19 @@ along with nucypher. If not, see <https://www.gnu.org/licenses/>.
"""
from typing import Optional, Dict, Tuple
from bytestring_splitter import BytestringSplitter
from eth_typing.evm import ChecksumAddress
from eth_utils.address import to_canonical_address, to_checksum_address
from nucypher.core import MessageKit, AuthorizedKeyFrag
from nucypher.crypto.signing import SignatureStamp, InvalidSignature
from nucypher.crypto.splitters import signature_splitter, checksum_address_splitter
from nucypher.crypto.umbral_adapter import Signature, PublicKey
from nucypher.utilities.versioning import Versioned
class RevocationOrder(Versioned):
"""
Represents a string used by characters to perform a revocation on a specific
Ursula. It's a bytestring made of the following format:
REVOKE-<arrangement id to revoke><signature of the previous string>
This is sent as a payload in a DELETE method to the /KFrag/ endpoint.
"""
def __init__(self,
ursula_checksum_address: ChecksumAddress,
# TODO: Use staker address instead (what if the staker rebonds)?
encrypted_kfrag: MessageKit,
signer: Optional[SignatureStamp] = None,
signature: Optional[Signature] = None):
self.ursula_checksum_address = ursula_checksum_address
self.encrypted_kfrag = encrypted_kfrag
if not (bool(signer) ^ bool(signature)):
raise ValueError("Either pass a signer or a signature; not both.")
elif signer:
self.signature = signer(self._body())
elif signature:
self.signature = signature
def __repr__(self):
return bytes(self)
def __len__(self):
return len(bytes(self))
def __eq__(self, other):
return bytes(self) == bytes(other)
def verify_signature(self, alice_verifying_key: PublicKey) -> bool:
"""
Verifies the revocation was from the provided pubkey.
"""
if not self.signature.verify(self._body(), alice_verifying_key):
raise InvalidSignature(f"Revocation has an invalid signature: {self.signature}")
return True
@classmethod
def _brand(cls) -> bytes:
return b'Revo'
@classmethod
def _version(cls) -> Tuple[int, int]:
return 1, 0
@classmethod
def _old_version_handlers(cls) -> Dict:
return {}
def _body(self) -> bytes:
return to_canonical_address(self.ursula_checksum_address) + bytes(self.encrypted_kfrag)
def _payload(self) -> bytes:
return self._body() + bytes(self.signature)
@classmethod
def _from_bytes_current(cls, data):
splitter = BytestringSplitter(
checksum_address_splitter, # ursula canonical address
(bytes, Versioned._HEADER_SIZE+AuthorizedKeyFrag.SERIALIZED_SIZE), # MessageKit version header + versioned ekfrag
signature_splitter
)
ursula_canonical_address, ekfrag, signature = splitter(data)
ursula_checksum_address = to_checksum_address(ursula_canonical_address)
return cls(ursula_checksum_address=ursula_checksum_address,
encrypted_kfrag=ekfrag,
signature=signature)
from nucypher.core import RevocationOrder
from nucypher.crypto.signing import SignatureStamp
class RevocationKit:
def __init__(self, treasure_map, signer: SignatureStamp):
# TODO: move to core and make a method of TreasureMap?
self.revocations = dict()
for node_id, encrypted_kfrag in treasure_map:
self.revocations[node_id] = RevocationOrder(ursula_checksum_address=node_id,
encrypted_kfrag=encrypted_kfrag,
signer=signer)
self.revocations[node_id] = RevocationOrder.author(ursula_address=node_id,
encrypted_kfrag=encrypted_kfrag,
signer=signer.as_umbral_signer())
def __iter__(self):
return iter(self.revocations.values())

3
tests/integration/characters/test_federated_grant_and_revoke.py
View File

@ -21,11 +21,10 @@ import datetime
import maya
import pytest
from nucypher.core import MessageKit
from nucypher.core import MessageKit, RevocationOrder
from nucypher.characters.lawful import Enrico
from nucypher.crypto.utils import keccak_digest
from nucypher.policy.revocation import RevocationOrder
def test_federated_grant(federated_alice, federated_bob, federated_ursulas):