Fix jsonld signatures

Also some schemas are no longer remotely accessed
main
Bob Mottram 2021-01-04 19:02:24 +00:00
parent 6aa1cc8389
commit 9bdfec94f0
7 changed files with 610 additions and 189 deletions

View File

@ -13235,6 +13235,12 @@ class PubServer(BaseHTTPRequestHandler):
if self.server.debug: if self.server.debug:
print('DEBUG: Check message has params') print('DEBUG: Check message has params')
if not messageJson:
self.send_response(403)
self.end_headers()
self.server.POSTbusy = False
return
if self.path.endswith('/inbox') or \ if self.path.endswith('/inbox') or \
self.path == '/sharedInbox': self.path == '/sharedInbox':
if not inboxMessageHasParams(messageJson): if not inboxMessageHasParams(messageJson):

View File

@ -10,7 +10,7 @@ import json
import os import os
import datetime import datetime
import time import time
from jsonldsig import jsonldVerify from linked_data_sig import verifyJsonSignature
from utils import hasUsersPath from utils import hasUsersPath
from utils import validPostDate from utils import validPostDate
from utils import getFullDomain from utils import getFullDomain
@ -2705,28 +2705,29 @@ def runInboxQueue(recentPostsCache: {}, maxRecentPosts: int,
if debug: if debug:
print('DEBUG: http header signature check success') print('DEBUG: http header signature check success')
# check json signature # check if a json signature exists on this post
checkJsonSignature = False checkJsonSignature = False
if queueJson['original'].get('@context') and \ if queueJson['original'].get('@context') and \
queueJson['original'].get('signature'): queueJson['original'].get('signature'):
if isinstance(queueJson['original']['signature'], dict): if isinstance(queueJson['original']['signature'], dict):
if queueJson['original']['signature'].get('type') and \ # see https://tools.ietf.org/html/rfc7515
queueJson['original']['signature'].get('signatureValue'): jwebsig = queueJson['original']['signature']
if queueJson['original']['signature']['type'] == \ # signature exists and is of the expected type
'RsaSignature2017': if jwebsig.get('type') and jwebsig.get('signatureValue'):
if jwebsig['type'] == 'RsaSignature2017':
checkJsonSignature = True checkJsonSignature = True
if checkJsonSignature: if checkJsonSignature:
# use the original json message received, not one which may have # use the original json message received, not one which may have
# been modified along the way # been modified along the way
if not jsonldVerify(queueJson['original'], pubKey): if not verifyJsonSignature(queueJson['original'], pubKey):
print('WARN: jsonld inbox signature check failed ' + print('WARN: jsonld inbox signature check failed ' +
keyId + ' ' + pubKey + ' ' + keyId + ' ' + pubKey + ' ' +
str(queueJson['original'])) str(queueJson['original']))
# if os.path.isfile(queueFilename): if os.path.isfile(queueFilename):
# os.remove(queueFilename) os.remove(queueFilename)
# if len(queue) > 0: if len(queue) > 0:
# queue.pop(0) queue.pop(0)
# continue continue
else: else:
print('jsonld inbox signature check success') print('jsonld inbox signature check success')

View File

@ -1,167 +0,0 @@
__filename__ = "jsonldsig.py"
__author__ = "Bob Mottram"
__credits__ = ['Based on ' +
'https://github.com/WebOfTrustInfo/ld-signatures-python']
__license__ = "AGPL3+"
__version__ = "1.1.0"
__maintainer__ = "Bob Mottram"
__email__ = "bob@freedombone.net"
__status__ = "Production"
from copy import deepcopy
from datetime import datetime
import pytz
try:
from Cryptodome.PublicKey import RSA
from Cryptodome.Hash import SHA256
from Cryptodome.Signature import pkcs1_5 as PKCS1_v1_5
except ImportError:
from Crypto.PublicKey import RSA
from Crypto.Hash import SHA256
from Crypto.Signature import PKCS1_v1_5
from pyjsonld import normalize
import base64
import json
def _b64safeEncode(payload: {}) -> str:
"""
b64 url safe encoding with the padding removed.
"""
return base64.urlsafe_b64encode(payload).rstrip(b'=')
def _b64safeDecode(payload: {}) -> str:
"""
b64 url safe decoding with the padding added.
"""
return base64.urlsafe_b64decode(payload + b'=' * (4 - len(payload) % 4))
def _normalizeJson(payload: {}) -> str:
"""
Normalize with URDNA2015
"""
return json.dumps(payload, separators=(',', ':'),
sort_keys=True).encode('utf-8')
def _signRs256(payload: {}, privateKeyPem: str) -> str:
"""
Produce a RS256 signature of the payload
"""
key = RSA.importKey(privateKeyPem)
signer = PKCS1_v1_5.new(key)
signature = signer.sign(SHA256.new(payload))
return signature
def _verifyRs256(payload: {}, signature: str, publicKeyPem: str) -> bool:
"""
Verifies a RS256 signature
"""
key = RSA.importKey(publicKeyPem)
verifier = PKCS1_v1_5.new(key)
return verifier.verify(SHA256.new(payload), signature)
def _signJws(payload: {}, privateKeyPem: str) -> str:
"""
Prepare payload to sign
"""
header = {
'alg': 'RS256',
'b64': False,
'crit': ['b64']
}
normalizedJson = _normalizeJson(header)
encodedHeader = _b64safeEncode(normalizedJson)
preparedPayload = b'.'.join([encodedHeader, payload])
signature = _signRs256(preparedPayload, privateKeyPem)
encodedSignature = _b64safeEncode(signature)
jwsSignature = b'..'.join([encodedHeader, encodedSignature])
return jwsSignature
def _verifyJws(payload: {}, jwsSignature: str, publicKeyPem: str) -> bool:
"""
Verifies a signature using the given public key
"""
if b'..' in jwsSignature:
encodedHeader, encodedSignature = jwsSignature.split(b'..')
signature = _b64safeDecode(encodedSignature)
payload = b'.'.join([encodedHeader, payload])
else:
signature = _b64safeDecode(jwsSignature)
payload = b'.'.join([payload])
return _verifyRs256(payload, signature, publicKeyPem)
def _jsonldNormalize(jldDocument: str):
"""
Normalize and hash the json-ld document
"""
options = {
'algorithm': 'URDNA2015',
'format': 'application/nquads'
}
normalized = normalize(jldDocument, options=options)
normalizedHash = SHA256.new(data=normalized.encode('utf-8')).digest()
return normalizedHash
def jsonldSign(jldDocument: {}, privateKeyPem: str) -> {}:
"""
Produces a signed JSON-LD document with a Json Web Signature
"""
if not jldDocument.get('@context'):
print('WARN: json document must have @context to sign')
return jldDocument
jldDocument = deepcopy(jldDocument)
normalizedJldHash = _jsonldNormalize(jldDocument)
jwsSignature = _signJws(normalizedJldHash, privateKeyPem)
# construct the signature document and add it to jsonld
signature = {
'type': 'RsaSignatureSuite2017',
'created': datetime.now(tz=pytz.utc).strftime('%Y-%m-%dT%H:%M:%SZ'),
'signatureValue': jwsSignature.decode('utf-8')
}
jldDocument.update({'signature': signature})
return jldDocument
def jsonldVerify(signedJldDocument: {}, publicKeyPem: str) -> bool:
"""
Verifies the Json Web Signature of a signed JSON-LD Document
"""
if not isinstance(signedJldDocument, dict):
return False
if not signedJldDocument.get('@context'):
print('json document must have @context')
return False
signedJldDocument = deepcopy(signedJldDocument)
signature = signedJldDocument.pop('signature')
jwsSignature = signature['signatureValue'].encode('utf-8')
normalizedJldHash = _jsonldNormalize(signedJldDocument)
return _verifyJws(normalizedJldHash, jwsSignature, publicKeyPem)
def testSignJsonld(jldDocument: {}, privateKeyPem: str) -> {}:
"""
Creates a test signature
"""
signedJldDocument = jsonldSign(jldDocument, privateKeyPem)
# pop the created time key since its dynamic
signedJldDocument['signature'].pop('created')
return signedJldDocument

84
linked_data_sig.py 100644
View File

@ -0,0 +1,84 @@
__filename__ = "linked_data_sig.py"
__author__ = "Bob Mottram"
__credits__ = ['Based on ' +
'https://github.com/tsileo/little-boxes']
__license__ = "AGPL3+"
__version__ = "1.1.0"
__maintainer__ = "Bob Mottram"
__email__ = "bob@freedombone.net"
__status__ = "Production"
import base64
import hashlib
from datetime import datetime
try:
from Cryptodome.PublicKey import RSA
from Cryptodome.Hash import SHA256
from Cryptodome.Signature import pkcs1_5 as PKCS1_v1_5
except ImportError:
from Crypto.PublicKey import RSA
from Crypto.Hash import SHA256
from Crypto.Signature import PKCS1_v1_5
from pyjsonld import normalize
def _options_hash(doc):
doc = dict(doc["signature"])
for k in ["type", "id", "signatureValue"]:
if k in doc:
del doc[k]
doc["@context"] = "https://w3id.org/identity/v1"
options = {
"algorithm": "URDNA2015",
"format": "application/nquads"
}
normalized = normalize(doc, options)
h = hashlib.new("sha256")
h.update(normalized.encode("utf-8"))
return h.hexdigest()
def _doc_hash(doc):
doc = dict(doc)
if "signature" in doc:
del doc["signature"]
options = {
"algorithm": "URDNA2015",
"format": "application/nquads"
}
normalized = normalize(doc, options)
h = hashlib.new("sha256")
h.update(normalized.encode("utf-8"))
return h.hexdigest()
def verifyJsonSignature(doc: {}, publicKeyPem: str):
key = RSA.importKey(publicKeyPem)
to_be_signed = _options_hash(doc) + _doc_hash(doc)
signature = doc["signature"]["signatureValue"]
signer = PKCS1_v1_5.new(key) # type: ignore
digest = SHA256.new()
digest.update(to_be_signed.encode("utf-8"))
return signer.verify(digest, base64.b64decode(signature)) # type: ignore
def generateJsonSignature(doc: {}, privateKeyPem: str):
if not doc.get('actor'):
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 = RSA.importKey(privateKeyPem)
signer = PKCS1_v1_5.new(key)
digest = SHA256.new()
digest.update(to_be_signed.encode("utf-8"))
sig = base64.b64encode(signer.sign(digest)) # type: ignore
options["signatureValue"] = sig.decode("utf-8")

View File

@ -65,7 +65,7 @@ from blocking import isBlocked
from blocking import isBlockedDomain from blocking import isBlockedDomain
from filters import isFiltered from filters import isFiltered
from git import convertPostToPatch from git import convertPostToPatch
from jsonldsig import jsonldSign from linked_data_sig import generateJsonSignature
from petnames import resolvePetnames from petnames import resolvePetnames
@ -1794,7 +1794,8 @@ def sendPost(projectVersion: str,
if not postJsonObject.get('signature'): if not postJsonObject.get('signature'):
try: try:
signedPostJsonObject = jsonldSign(postJsonObject, privateKeyPem) signedPostJsonObject = postJsonObject.copy()
generateJsonSignature(signedPostJsonObject, privateKeyPem)
postJsonObject = signedPostJsonObject postJsonObject = signedPostJsonObject
except Exception as e: except Exception as e:
print('WARN: failed to JSON-LD sign post, ' + str(e)) print('WARN: failed to JSON-LD sign post, ' + str(e))
@ -2122,7 +2123,8 @@ def sendSignedJson(postJsonObject: {}, session, baseDir: str,
if not postJsonObject.get('signature'): if not postJsonObject.get('signature'):
try: try:
signedPostJsonObject = jsonldSign(postJsonObject, privateKeyPem) signedPostJsonObject = postJsonObject.copy()
generateJsonSignature(signedPostJsonObject, privateKeyPem)
postJsonObject = signedPostJsonObject postJsonObject = signedPostJsonObject
except Exception as e: except Exception as e:
print('WARN: failed to JSON-LD sign post, ' + str(e)) print('WARN: failed to JSON-LD sign post, ' + str(e))

View File

@ -355,6 +355,476 @@ def parse_link_header(header):
return rval return rval
def getV1Schema() -> {}:
# https://w3id.org/identity/v1
return {
"@context": {
"id": "@id",
"type": "@type",
"cred": "https://w3id.org/credentials#",
"dc": "http://purl.org/dc/terms/",
"identity": "https://w3id.org/identity#",
"perm": "https://w3id.org/permissions#",
"ps": "https://w3id.org/payswarm#",
"rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
"rdfs": "http://www.w3.org/2000/01/rdf-schema#",
"sec": "https://w3id.org/security#",
"schema": "http://schema.org/",
"xsd": "http://www.w3.org/2001/XMLSchema#",
"Group": "https://www.w3.org/ns/activitystreams#Group",
"claim": {"@id": "cred:claim", "@type": "@id"},
"credential": {"@id": "cred:credential", "@type": "@id"},
"issued": {"@id": "cred:issued", "@type": "xsd:dateTime"},
"issuer": {"@id": "cred:issuer", "@type": "@id"},
"recipient": {"@id": "cred:recipient", "@type": "@id"},
"Credential": "cred:Credential",
"CryptographicKeyCredential": "cred:CryptographicKeyCredential",
"about": {"@id": "schema:about", "@type": "@id"},
"address": {"@id": "schema:address", "@type": "@id"},
"addressCountry": "schema:addressCountry",
"addressLocality": "schema:addressLocality",
"addressRegion": "schema:addressRegion",
"comment": "rdfs:comment",
"created": {"@id": "dc:created", "@type": "xsd:dateTime"},
"creator": {"@id": "dc:creator", "@type": "@id"},
"description": "schema:description",
"email": "schema:email",
"familyName": "schema:familyName",
"givenName": "schema:givenName",
"image": {"@id": "schema:image", "@type": "@id"},
"label": "rdfs:label",
"name": "schema:name",
"postalCode": "schema:postalCode",
"streetAddress": "schema:streetAddress",
"title": "dc:title",
"url": {"@id": "schema:url", "@type": "@id"},
"Person": "schema:Person",
"PostalAddress": "schema:PostalAddress",
"Organization": "schema:Organization",
"identityService": {
"@id": "identity:identityService", "@type": "@id"
},
"idp": {"@id": "identity:idp", "@type": "@id"},
"Identity": "identity:Identity",
"paymentProcessor": "ps:processor",
"preferences": {"@id": "ps:preferences", "@type": "@vocab"},
"cipherAlgorithm": "sec:cipherAlgorithm",
"cipherData": "sec:cipherData",
"cipherKey": "sec:cipherKey",
"digestAlgorithm": "sec:digestAlgorithm",
"digestValue": "sec:digestValue",
"domain": "sec:domain",
"expires": {"@id": "sec:expiration", "@type": "xsd:dateTime"},
"initializationVector": "sec:initializationVector",
"member": {"@id": "schema:member", "@type": "@id"},
"memberOf": {"@id": "schema:memberOf", "@type": "@id"},
"nonce": "sec:nonce",
"normalizationAlgorithm": "sec:normalizationAlgorithm",
"owner": {"@id": "sec:owner", "@type": "@id"},
"password": "sec:password",
"privateKey": {"@id": "sec:privateKey", "@type": "@id"},
"privateKeyPem": "sec:privateKeyPem",
"publicKey": {"@id": "sec:publicKey", "@type": "@id"},
"publicKeyPem": "sec:publicKeyPem",
"publicKeyService": {
"@id": "sec:publicKeyService", "@type": "@id"
},
"revoked": {"@id": "sec:revoked", "@type": "xsd:dateTime"},
"signature": "sec:signature",
"signatureAlgorithm": "sec:signatureAlgorithm",
"signatureValue": "sec:signatureValue",
"CryptographicKey": "sec:Key",
"EncryptedMessage": "sec:EncryptedMessage",
"GraphSignature2012": "sec:GraphSignature2012",
"LinkedDataSignature2015": "sec:LinkedDataSignature2015",
"accessControl": {"@id": "perm:accessControl", "@type": "@id"},
"writePermission": {"@id": "perm:writePermission", "@type": "@id"}
}
}
def getActivitystreamsSchema() -> {}:
return {
"@context": {
"@vocab": "_:",
"xsd": "http://www.w3.org/2001/XMLSchema#",
"as": "https://www.w3.org/ns/activitystreams#",
"ldp": "http://www.w3.org/ns/ldp#",
"vcard": "http://www.w3.org/2006/vcard/ns#",
"id": "@id",
"type": "@type",
"Accept": "as:Accept",
"Activity": "as:Activity",
"IntransitiveActivity": "as:IntransitiveActivity",
"Add": "as:Add",
"Announce": "as:Announce",
"Application": "as:Application",
"Arrive": "as:Arrive",
"Article": "as:Article",
"Audio": "as:Audio",
"Block": "as:Block",
"Collection": "as:Collection",
"CollectionPage": "as:CollectionPage",
"Relationship": "as:Relationship",
"Create": "as:Create",
"Delete": "as:Delete",
"Dislike": "as:Dislike",
"Document": "as:Document",
"Event": "as:Event",
"Follow": "as:Follow",
"Flag": "as:Flag",
"Group": "as:Group",
"Ignore": "as:Ignore",
"Image": "as:Image",
"Invite": "as:Invite",
"Join": "as:Join",
"Leave": "as:Leave",
"Like": "as:Like",
"Link": "as:Link",
"Mention": "as:Mention",
"Note": "as:Note",
"Object": "as:Object",
"Offer": "as:Offer",
"OrderedCollection": "as:OrderedCollection",
"OrderedCollectionPage": "as:OrderedCollectionPage",
"Organization": "as:Organization",
"Page": "as:Page",
"Person": "as:Person",
"Place": "as:Place",
"Profile": "as:Profile",
"Question": "as:Question",
"Reject": "as:Reject",
"Remove": "as:Remove",
"Service": "as:Service",
"TentativeAccept": "as:TentativeAccept",
"TentativeReject": "as:TentativeReject",
"Tombstone": "as:Tombstone",
"Undo": "as:Undo",
"Update": "as:Update",
"Video": "as:Video",
"View": "as:View",
"Listen": "as:Listen",
"Read": "as:Read",
"Move": "as:Move",
"Travel": "as:Travel",
"IsFollowing": "as:IsFollowing",
"IsFollowedBy": "as:IsFollowedBy",
"IsContact": "as:IsContact",
"IsMember": "as:IsMember",
"subject": {
"@id": "as:subject",
"@type": "@id"
},
"relationship": {
"@id": "as:relationship",
"@type": "@id"
},
"actor": {
"@id": "as:actor",
"@type": "@id"
},
"attributedTo": {
"@id": "as:attributedTo",
"@type": "@id"
},
"attachment": {
"@id": "as:attachment",
"@type": "@id"
},
"bcc": {
"@id": "as:bcc",
"@type": "@id"
},
"bto": {
"@id": "as:bto",
"@type": "@id"
},
"cc": {
"@id": "as:cc",
"@type": "@id"
},
"context": {
"@id": "as:context",
"@type": "@id"
},
"current": {
"@id": "as:current",
"@type": "@id"
},
"first": {
"@id": "as:first",
"@type": "@id"
},
"generator": {
"@id": "as:generator",
"@type": "@id"
},
"icon": {
"@id": "as:icon",
"@type": "@id"
},
"image": {
"@id": "as:image",
"@type": "@id"
},
"inReplyTo": {
"@id": "as:inReplyTo",
"@type": "@id"
},
"items": {
"@id": "as:items",
"@type": "@id"
},
"instrument": {
"@id": "as:instrument",
"@type": "@id"
},
"orderedItems": {
"@id": "as:items",
"@type": "@id",
"@container": "@list"
},
"last": {
"@id": "as:last",
"@type": "@id"
},
"location": {
"@id": "as:location",
"@type": "@id"
},
"next": {
"@id": "as:next",
"@type": "@id"
},
"object": {
"@id": "as:object",
"@type": "@id"
},
"oneOf": {
"@id": "as:oneOf",
"@type": "@id"
},
"anyOf": {
"@id": "as:anyOf",
"@type": "@id"
},
"closed": {
"@id": "as:closed",
"@type": "xsd:dateTime"
},
"origin": {
"@id": "as:origin",
"@type": "@id"
},
"accuracy": {
"@id": "as:accuracy",
"@type": "xsd:float"
},
"prev": {
"@id": "as:prev",
"@type": "@id"
},
"preview": {
"@id": "as:preview",
"@type": "@id"
},
"replies": {
"@id": "as:replies",
"@type": "@id"
},
"result": {
"@id": "as:result",
"@type": "@id"
},
"audience": {
"@id": "as:audience",
"@type": "@id"
},
"partOf": {
"@id": "as:partOf",
"@type": "@id"
},
"tag": {
"@id": "as:tag",
"@type": "@id"
},
"target": {
"@id": "as:target",
"@type": "@id"
},
"to": {
"@id": "as:to",
"@type": "@id"
},
"url": {
"@id": "as:url",
"@type": "@id"
},
"altitude": {
"@id": "as:altitude",
"@type": "xsd:float"
},
"content": "as:content",
"contentMap": {
"@id": "as:content",
"@container": "@language"
},
"name": "as:name",
"nameMap": {
"@id": "as:name",
"@container": "@language"
},
"duration": {
"@id": "as:duration",
"@type": "xsd:duration"
},
"endTime": {
"@id": "as:endTime",
"@type": "xsd:dateTime"
},
"height": {
"@id": "as:height",
"@type": "xsd:nonNegativeInteger"
},
"href": {
"@id": "as:href",
"@type": "@id"
},
"hreflang": "as:hreflang",
"latitude": {
"@id": "as:latitude",
"@type": "xsd:float"
},
"longitude": {
"@id": "as:longitude",
"@type": "xsd:float"
},
"mediaType": "as:mediaType",
"published": {
"@id": "as:published",
"@type": "xsd:dateTime"
},
"radius": {
"@id": "as:radius",
"@type": "xsd:float"
},
"rel": "as:rel",
"startIndex": {
"@id": "as:startIndex",
"@type": "xsd:nonNegativeInteger"
},
"startTime": {
"@id": "as:startTime",
"@type": "xsd:dateTime"
},
"summary": "as:summary",
"summaryMap": {
"@id": "as:summary",
"@container": "@language"
},
"totalItems": {
"@id": "as:totalItems",
"@type": "xsd:nonNegativeInteger"
},
"units": "as:units",
"updated": {
"@id": "as:updated",
"@type": "xsd:dateTime"
},
"width": {
"@id": "as:width",
"@type": "xsd:nonNegativeInteger"
},
"describes": {
"@id": "as:describes",
"@type": "@id"
},
"formerType": {
"@id": "as:formerType",
"@type": "@id"
},
"deleted": {
"@id": "as:deleted",
"@type": "xsd:dateTime"
},
"inbox": {
"@id": "ldp:inbox",
"@type": "@id"
},
"outbox": {
"@id": "as:outbox",
"@type": "@id"
},
"following": {
"@id": "as:following",
"@type": "@id"
},
"followers": {
"@id": "as:followers",
"@type": "@id"
},
"streams": {
"@id": "as:streams",
"@type": "@id"
},
"preferredUsername": "as:preferredUsername",
"endpoints": {
"@id": "as:endpoints",
"@type": "@id"
},
"uploadMedia": {
"@id": "as:uploadMedia",
"@type": "@id"
},
"proxyUrl": {
"@id": "as:proxyUrl",
"@type": "@id"
},
"liked": {
"@id": "as:liked",
"@type": "@id"
},
"oauthAuthorizationEndpoint": {
"@id": "as:oauthAuthorizationEndpoint",
"@type": "@id"
},
"oauthTokenEndpoint": {
"@id": "as:oauthTokenEndpoint",
"@type": "@id"
},
"provideClientKey": {
"@id": "as:provideClientKey",
"@type": "@id"
},
"signClientKey": {
"@id": "as:signClientKey",
"@type": "@id"
},
"sharedInbox": {
"@id": "as:sharedInbox",
"@type": "@id"
},
"Public": {
"@id": "as:Public",
"@type": "@id"
},
"source": "as:source",
"likes": {
"@id": "as:likes",
"@type": "@id"
},
"shares": {
"@id": "as:shares",
"@type": "@id"
},
"alsoKnownAs": {
"@id": "as:alsoKnownAs",
"@type": "@id"
}
}
}
def load_document(url): def load_document(url):
""" """
Retrieves JSON-LD at the given URL. Retrieves JSON-LD at the given URL.
@ -380,6 +850,22 @@ def load_document(url):
url_opener.addheaders = [ url_opener.addheaders = [
('Accept', 'application/ld+json, application/json'), ('Accept', 'application/ld+json, application/json'),
('Accept-Encoding', 'deflate')] ('Accept-Encoding', 'deflate')]
if url == 'https://w3id.org/identity/v1':
doc = {
'contextUrl': None,
'documentUrl': url,
'document': getV1Schema()
}
return doc
elif url == 'https://www.w3.org/ns/activitystreams':
doc = {
'contextUrl': None,
'documentUrl': url,
'document': getActivitystreamsSchema()
}
return doc
with closing(url_opener.open(url)) as handle: with closing(url_opener.open(url)) as handle:
if handle.info().get('Content-Encoding') == 'gzip': if handle.info().get('Content-Encoding') == 'gzip':
buf = io.BytesIO(handle.read()) buf = io.BytesIO(handle.read())

View File

@ -86,8 +86,8 @@ from content import replaceContentDuplicates
from content import removeTextFormatting from content import removeTextFormatting
from content import removeHtmlTag from content import removeHtmlTag
from theme import setCSSparam from theme import setCSSparam
from jsonldsig import testSignJsonld from linked_data_sig import generateJsonSignature
from jsonldsig import jsonldVerify from linked_data_sig import verifyJsonSignature
from newsdaemon import hashtagRuleTree from newsdaemon import hashtagRuleTree
from newsdaemon import hashtagRuleResolve from newsdaemon import hashtagRuleResolve
from newswire import getNewswireTags from newswire import getNewswireTags
@ -1977,8 +1977,10 @@ def testRemoveTextFormatting():
def testJsonld(): def testJsonld():
print("testJsonld") print("testJsonld")
jldDocument = { jldDocument = {
"@context": "https://www.w3.org/ns/activitystreams", "@context": "https://www.w3.org/ns/activitystreams",
"actor": "https://somesite.net/users/gerbil",
"description": "My json document", "description": "My json document",
"numberField": 83582, "numberField": 83582,
"object": { "object": {
@ -2023,27 +2025,32 @@ def testJsonld():
'TwIDAQAB\n' \ 'TwIDAQAB\n' \
'-----END PUBLIC KEY-----' '-----END PUBLIC KEY-----'
signedDocument = testSignJsonld(jldDocument, privateKeyPem) signedDocument = jldDocument.copy()
generateJsonSignature(signedDocument, privateKeyPem)
assert(signedDocument) assert(signedDocument)
assert(signedDocument.get('signature')) assert(signedDocument.get('signature'))
assert(signedDocument['signature'].get('signatureValue')) assert(signedDocument['signature'].get('signatureValue'))
assert(signedDocument['signature'].get('type')) assert(signedDocument['signature'].get('type'))
assert(len(signedDocument['signature']['signatureValue']) > 50) assert(len(signedDocument['signature']['signatureValue']) > 50)
assert(signedDocument['signature']['type'] == 'RsaSignatureSuite2017') # print(str(signedDocument['signature']))
assert(jsonldVerify(signedDocument, publicKeyPem)) assert(signedDocument['signature']['type'] == 'RsaSignature2017')
assert(verifyJsonSignature(signedDocument, publicKeyPem))
# alter the signed document # alter the signed document
signedDocument['object']['content'] = 'forged content' signedDocument['object']['content'] = 'forged content'
assert(not jsonldVerify(signedDocument, publicKeyPem)) assert(not verifyJsonSignature(signedDocument, publicKeyPem))
jldDocument2 = { jldDocument2 = {
"@context": "https://www.w3.org/ns/activitystreams", "@context": "https://www.w3.org/ns/activitystreams",
"actor": "https://somesite.net/users/gerbil",
"description": "Another json document", "description": "Another json document",
"numberField": 13353, "numberField": 13353,
"object": { "object": {
"content": "More content" "content": "More content"
} }
} }
signedDocument2 = testSignJsonld(jldDocument2, privateKeyPem) signedDocument2 = jldDocument2.copy()
generateJsonSignature(signedDocument2, privateKeyPem)
assert(signedDocument2) assert(signedDocument2)
assert(signedDocument2.get('signature')) assert(signedDocument2.get('signature'))
assert(signedDocument2['signature'].get('signatureValue')) assert(signedDocument2['signature'].get('signatureValue'))
@ -2051,6 +2058,8 @@ def testJsonld():
if signedDocument['signature']['signatureValue'] == \ if signedDocument['signature']['signatureValue'] == \
signedDocument2['signature']['signatureValue']: signedDocument2['signature']['signatureValue']:
print('json signature has not changed for different documents') print('json signature has not changed for different documents')
assert '.' not in str(signedDocument['signature']['signatureValue'])
assert len(str(signedDocument['signature']['signatureValue'])) > 340
assert(signedDocument['signature']['signatureValue'] != assert(signedDocument['signature']['signatureValue'] !=
signedDocument2['signature']['signatureValue']) signedDocument2['signature']['signatureValue'])