mirror of https://gitlab.com/bashrc2/epicyon
117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
__filename__ = "linked_data_sig.py"
|
|
__author__ = "Bob Mottram"
|
|
__credits__ = ['Based on ' +
|
|
'https://github.com/tsileo/little-boxes']
|
|
__license__ = "AGPL3+"
|
|
__version__ = "1.3.0"
|
|
__maintainer__ = "Bob Mottram"
|
|
__email__ = "bob@libreserver.org"
|
|
__status__ = "Production"
|
|
__module_group__ = "Security"
|
|
|
|
import base64
|
|
import hashlib
|
|
from datetime import datetime
|
|
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.hazmat.primitives.serialization import load_pem_private_key
|
|
from cryptography.hazmat.primitives.serialization import load_pem_public_key
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.asymmetric import utils as hazutils
|
|
from pyjsonld import normalize
|
|
from context import has_valid_context
|
|
from utils import get_sha_256
|
|
|
|
|
|
def _options_hash(doc: {}) -> str:
|
|
"""Returns a hash of the signature, with a few fields removed
|
|
"""
|
|
doc_sig = dict(doc["signature"])
|
|
|
|
# remove fields from signature
|
|
for key in ["type", "id", "signatureValue"]:
|
|
if key in doc_sig:
|
|
del doc_sig[key]
|
|
|
|
doc_sig["@context"] = "https://w3id.org/identity/v1"
|
|
options = {
|
|
"algorithm": "URDNA2015",
|
|
"format": "application/nquads"
|
|
}
|
|
|
|
normalized = normalize(doc_sig, options)
|
|
hsh = hashlib.new("sha256")
|
|
hsh.update(normalized.encode("utf-8"))
|
|
return hsh.hexdigest()
|
|
|
|
|
|
def _doc_hash(doc: {}) -> str:
|
|
"""Returns a hash of the ActivityPub post
|
|
"""
|
|
doc = dict(doc)
|
|
|
|
# remove the signature
|
|
if "signature" in doc:
|
|
del doc["signature"]
|
|
|
|
options = {
|
|
"algorithm": "URDNA2015",
|
|
"format": "application/nquads"
|
|
}
|
|
|
|
normalized = normalize(doc, options)
|
|
hsh = hashlib.new("sha256")
|
|
hsh.update(normalized.encode("utf-8"))
|
|
return hsh.hexdigest()
|
|
|
|
|
|
def verify_json_signature(doc: {}, public_key_pem: str) -> bool:
|
|
"""Returns True if the given ActivityPub post was sent
|
|
by an actor having the given public key
|
|
"""
|
|
if not has_valid_context(doc):
|
|
return False
|
|
pubkey = load_pem_public_key(public_key_pem.encode('utf-8'),
|
|
backend=default_backend())
|
|
to_be_signed = _options_hash(doc) + _doc_hash(doc)
|
|
signature = doc["signature"]["signatureValue"]
|
|
|
|
digest = get_sha_256(to_be_signed.encode("utf-8"))
|
|
base64sig = base64.b64decode(signature)
|
|
|
|
try:
|
|
pubkey.verify(
|
|
base64sig,
|
|
digest,
|
|
padding.PKCS1v15(),
|
|
hazutils.Prehashed(hashes.SHA256()))
|
|
return True
|
|
except BaseException:
|
|
print('EX: verify_json_signature unable to verify')
|
|
return False
|
|
|
|
|
|
def generate_json_signature(doc: {}, private_key_pem: str) -> None:
|
|
"""Adds a json signature to the given ActivityPub post
|
|
"""
|
|
if not doc.get('actor'):
|
|
return
|
|
if not has_valid_context(doc):
|
|
return
|
|
options = {
|
|
"type": "RsaSignature2017",
|
|
"creator": doc["actor"] + "#main-key",
|
|
"created": datetime.utcnow().replace(microsecond=0).isoformat() + "Z",
|
|
}
|
|
doc["signature"] = options
|
|
to_be_signed = _options_hash(doc) + _doc_hash(doc)
|
|
|
|
key = load_pem_private_key(private_key_pem.encode('utf-8'),
|
|
None, backend=default_backend())
|
|
digest = get_sha_256(to_be_signed.encode("utf-8"))
|
|
signature = key.sign(digest,
|
|
padding.PKCS1v15(),
|
|
hazutils.Prehashed(hashes.SHA256()))
|
|
sig = base64.b64encode(signature)
|
|
options["signatureValue"] = sig.decode("utf-8")
|