ci: Add and configure mypy, and make it happy

This commit is contained in:
2023-11-01 07:22:29 +01:00
parent f2a54b5ce6
commit 1c1e57d868
13 changed files with 374 additions and 214 deletions

View File

@@ -26,8 +26,11 @@ from base64 import b64encode
from collections import OrderedDict, namedtuple
from datetime import datetime
from hashlib import sha256
from typing import Any, Dict, Optional
from nacl.signing import SigningKey, VerifyKey
from simplejson import dumps, loads
from typing_extensions import Self
from ssb.util import tag
@@ -38,7 +41,7 @@ class NoPrivateKeyException(Exception):
"""Exception to raise when a private key is not available"""
def to_ordered(data):
def to_ordered(data: Dict[str, Any]) -> OrderedDict[str, Any]:
"""Convert a dictionary to an ``OrderedDict``"""
smsg = OrderedMsg(**data)
@@ -46,7 +49,7 @@ def to_ordered(data):
return OrderedDict((k, getattr(smsg, k)) for k in smsg._fields)
def get_millis_1970():
def get_millis_1970() -> int:
"""Get the UNIX timestamp in milliseconds"""
return int(datetime.utcnow().timestamp() * 1000)
@@ -55,16 +58,16 @@ def get_millis_1970():
class Feed:
"""Base class for feeds"""
def __init__(self, public_key):
def __init__(self, public_key: VerifyKey) -> None:
self.public_key = public_key
@property
def id(self):
def id(self) -> str:
"""The identifier of the feed"""
return tag(self.public_key).decode("ascii")
def sign(self, msg):
def sign(self, msg: bytes) -> bytes:
"""Sign a message"""
raise NoPrivateKeyException("Cannot use remote identity to sign (no private key!)")
@@ -73,16 +76,20 @@ class Feed:
class LocalFeed(Feed):
"""Class representing a local feed"""
def __init__(self, private_key): # pylint: disable=super-init-not-called
def __init__(self, private_key: SigningKey) -> None: # pylint: disable=super-init-not-called
self.private_key = private_key
@property
def public_key(self):
def public_key(self) -> VerifyKey:
"""The public key of the feed"""
return self.private_key.verify_key
def sign(self, msg):
@public_key.setter
def public_key(self, key: VerifyKey) -> None:
raise TypeError("Can not set only the public key for a local feed")
def sign(self, msg: bytes) -> bytes:
"""Sign a message for this feed"""
return self.private_key.sign(msg).signature
@@ -92,25 +99,34 @@ class Message:
"""Base class for SSB messages"""
def __init__( # pylint: disable=too-many-arguments
self, feed, content, signature=None, sequence=1, timestamp=None, previous=None
self,
feed: Feed,
content: Dict[str, Any],
signature: Optional[str] = None,
sequence: int = 1,
timestamp: Optional[int] = None,
previous: Optional["Message"] = None,
):
self.feed = feed
self.content = content
if signature is None:
raise ValueError("signature can't be None")
self.signature = signature
self.previous = previous
self.timestamp = get_millis_1970() if timestamp is None else timestamp
if self.previous:
self.sequence = self.previous.sequence + 1
self.sequence: int = self.previous.sequence + 1
else:
self.sequence = sequence
self.timestamp = get_millis_1970() if timestamp is None else timestamp
self._check_signature()
def _check_signature(self) -> None:
if self.signature is None:
raise ValueError("signature can't be None")
@classmethod
def parse(cls, data, feed):
def parse(cls, data: bytes, feed: Feed) -> Self:
"""Parse raw message data"""
obj = loads(data, object_pairs_hook=OrderedDict)
@@ -118,12 +134,12 @@ class Message:
return msg
def serialize(self, add_signature=True):
def serialize(self, add_signature: bool = True) -> bytes:
"""Serialize the message"""
return dumps(self.to_dict(add_signature=add_signature), indent=2).encode("utf-8")
def to_dict(self, add_signature=True):
def to_dict(self, add_signature: bool = True) -> OrderedDict[str, Any]:
"""Convert the message to a dictionary"""
obj = to_ordered(
@@ -142,20 +158,21 @@ class Message:
return obj
def verify(self, signature):
def verify(self, signature: str) -> bool:
"""Verify the signature of the message"""
return self.signature == signature
@property
def hash(self):
def hash(self) -> str:
"""The cryptographic hash of the message"""
hash_ = sha256(self.serialize()).digest()
return b64encode(hash_).decode("ascii") + ".sha256"
@property
def key(self):
def key(self) -> str:
"""The key of the message"""
return "%" + self.hash
@@ -165,25 +182,21 @@ class LocalMessage(Message):
"""Class representing a local message"""
def __init__( # pylint: disable=too-many-arguments,super-init-not-called
self, feed, content, signature=None, sequence=1, timestamp=None, previous=None
self,
feed: LocalFeed,
content: Dict[str, Any],
signature: Optional[str] = None,
sequence: int = 1,
timestamp: Optional[int] = None,
previous: Optional["LocalMessage"] = None,
):
self.feed = feed
self.content = content
super().__init__(feed, content, signature=signature, sequence=sequence, timestamp=timestamp, previous=previous)
self.previous = previous
if self.previous:
self.sequence = self.previous.sequence + 1
else:
self.sequence = sequence
self.timestamp = get_millis_1970() if timestamp is None else timestamp
if signature is None:
def _check_signature(self) -> None:
if self.signature is None:
self.signature = self._sign()
else:
self.signature = signature
def _sign(self):
def _sign(self) -> str:
# ensure ordering of keys and indentation of 2 characters, like ssb-keys
data = self.serialize(add_signature=False)
return (b64encode(bytes(self.feed.sign(data))) + b".sig.ed25519").decode("ascii")