LibertaCasa/takahe
Archived
9
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add JSON-LD signatures and tests for sig stuff

Browse Source
This commit is contained in:
Andrew Godwin 2022-11-12 15:10:15 -07:00
parent 8fd5a9292c
commit dd4328ae52
12 changed files with 501 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
activities/models/fan_out.py
View File

@ -32,7 +32,8 @@ class FanOutStates(StateGraph):
await HttpSignature.signed_request(
uri=fan_out.identity.inbox_uri,
body=canonicalise(post.to_create_ap()),
identity=post.author,
private_key=post.author.public_key,
key_id=post.author.public_key_id,
)
return cls.sent

85
core/ld.py
View File

@ -229,6 +229,91 @@ schemas = {
}
},
},
"w3id.org/identity/v1": {
"contentType": "application/ld+json",
"documentUrl": "http://w3id.org/identity/v1",
"contextUrl": None,
"document": {
"@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"},
}
},
},
"*/schemas/litepub-0.1.jsonld": {
"contentType": "application/ld+json",
"documentUrl": "http://w3id.org/security/v1",

190
core/signatures.py
View File

@ -1,16 +1,33 @@
import base64
import json
from typing import TYPE_CHECKING, Dict, List, Literal, TypedDict
from typing import Dict, List, Literal, TypedDict
from urllib.parse import urlparse
import httpx
from cryptography.hazmat.primitives import hashes
from django.http import HttpRequest
from django.utils.http import http_date
from django.utils import timezone
from django.utils.http import http_date, parse_http_date
from OpenSSL import crypto
from pyld import jsonld
# Prevent a circular import
if TYPE_CHECKING:
from users.models import Identity
from core.ld import format_date
class VerificationError(BaseException):
"""
There was an error with verifying the signature
"""
pass
class VerificationFormatError(VerificationError):
"""
There was an error with the format of the signature (not if it is valid)
"""
pass
class HttpSignature:
@ -47,13 +64,13 @@ class HttpSignature:
return "\n".join(f"{name.lower()}: {value}" for name, value in headers.items())
@classmethod
def parse_signature(cls, signature: str) -> "SignatureDetails":
def parse_signature(cls, signature: str) -> "HttpSignatureDetails":
bits = {}
for item in signature.split(","):
name, value = item.split("=", 1)
value = value.strip('"')
bits[name.lower()] = value
signature_details: SignatureDetails = {
signature_details: HttpSignatureDetails = {
"headers": bits["headers"].split(),
"signature": base64.b64decode(bits["signature"]),
"algorithm": bits["algorithm"],
@ -62,7 +79,7 @@ class HttpSignature:
return signature_details
@classmethod
def compile_signature(cls, details: "SignatureDetails") -> str:
def compile_signature(cls, details: "HttpSignatureDetails") -> str:
value = f'keyId="{details["keyid"]}",headers="'
value += " ".join(h.lower() for h in details["headers"])
value += '",signature="'
@ -70,12 +87,66 @@ class HttpSignature:
value += f'",algorithm="{details["algorithm"]}"'
return value
@classmethod
def verify_signature(
cls,
signature: bytes,
cleartext: str,
public_key: str,
):
x509 = crypto.X509()
x509.set_pubkey(
crypto.load_publickey(
crypto.FILETYPE_PEM,
public_key.encode("ascii"),
)
)
try:
crypto.verify(x509, signature, cleartext.encode("ascii"), "sha256")
except crypto.Error:
raise VerificationError("Signature mismatch")
@classmethod
def verify_request(cls, request, public_key, skip_date=False):
"""
Verifies that the request has a valid signature for its body
"""
# Verify body digest
if "HTTP_DIGEST" in request.META:
expected_digest = HttpSignature.calculate_digest(request.body)
if request.META["HTTP_DIGEST"] != expected_digest:
print("Wrong digest")
raise VerificationFormatError("Digest is incorrect")
# Verify date header
if "HTTP_DATE" in request.META and not skip_date:
header_date = parse_http_date(request.META["HTTP_DATE"])
if abs(timezone.now().timestamp() - header_date) > 60:
print(
f"Date mismatch - they sent {header_date}, now is {timezone.now().timestamp()}"
)
raise VerificationFormatError("Date is too far away")
# Get the signature details
if "HTTP_SIGNATURE" not in request.META:
raise VerificationFormatError("No signature header present")
signature_details = cls.parse_signature(request.META["HTTP_SIGNATURE"])
# Reject unknown algorithms
if signature_details["algorithm"] != "rsa-sha256":
raise VerificationFormatError("Unknown signature algorithm")
# Create the signature payload
headers_string = cls.headers_from_request(request, signature_details["headers"])
cls.verify_signature(
signature_details["signature"],
headers_string,
public_key,
)
@classmethod
async def signed_request(
self,
uri: str,
body: Dict,
identity: "Identity",
private_key: str,
key_id: str,
content_type: str = "application/json",
method: Literal["post"] = "post",
):
@ -96,11 +167,20 @@ class HttpSignature:
signed_string = "\n".join(
f"{name.lower()}: {value}" for name, value in headers.items()
)
pkey = crypto.load_privatekey(
crypto.FILETYPE_PEM,
private_key.encode("ascii"),
)
signature = crypto.sign(
pkey,
signed_string.encode("ascii"),
"sha256",
)
headers["Signature"] = self.compile_signature(
{
"keyid": identity.key_id,
"keyid": key_id,
"headers": list(headers.keys()),
"signature": identity.sign(signed_string),
"signature": signature,
"algorithm": "rsa-sha256",
}
)
@ -120,8 +200,94 @@ class HttpSignature:
return response
class SignatureDetails(TypedDict):
class HttpSignatureDetails(TypedDict):
algorithm: str
headers: List[str]
signature: bytes
keyid: str
class LDSignature:
"""
Creates and verifies signatures of JSON-LD documents
"""
@classmethod
def verify_signature(cls, document: Dict, public_key: str) -> None:
"""
Verifies a document
"""
try:
# Strip out the signature from the incoming document
signature = document.pop("signature")
# Create the options document
options = {
"@context": "https://w3id.org/identity/v1",
"creator": signature["creator"],
"created": signature["created"],
}
except KeyError:
raise VerificationFormatError("Invalid signature section")
if signature["type"].lower() != "rsasignature2017":
raise VerificationFormatError("Unknown signature type")
# Get the normalised hash of each document
final_hash = cls.normalized_hash(options) + cls.normalized_hash(document)
# Verify the signature
x509 = crypto.X509()
x509.set_pubkey(
crypto.load_publickey(
crypto.FILETYPE_PEM,
public_key.encode("ascii"),
)
)
try:
crypto.verify(
x509,
base64.b64decode(signature["signatureValue"]),
final_hash,
"sha256",
)
except crypto.Error:
raise VerificationError("Signature mismatch")
@classmethod
def create_signature(
cls, document: Dict, private_key: str, key_id: str
) -> Dict[str, str]:
"""
Creates the signature for a document
"""
# Create the options document
options: Dict[str, str] = {
"@context": "https://w3id.org/identity/v1",
"creator": key_id,
"created": format_date(timezone.now()),
}
# Get the normalised hash of each document
final_hash = cls.normalized_hash(options) + cls.normalized_hash(document)
# Create the signature
pkey = crypto.load_privatekey(
crypto.FILETYPE_PEM,
private_key.encode("ascii"),
)
signature = base64.b64encode(crypto.sign(pkey, final_hash, "sha256"))
# Add it to the options document along with other bits
options["signatureValue"] = signature.decode("ascii")
options["type"] = "RsaSignature2017"
return options
@classmethod
def normalized_hash(cls, document) -> bytes:
"""
Takes a JSON-LD document and create a hash of its URDNA2015 form,
in the same way that Mastodon does internally.
Reference: https://socialhub.activitypub.rocks/t/making-sense-of-rsasignature2017/347
"""
norm_form = jsonld.normalize(
document,
{"algorithm": "URDNA2015", "format": "application/n-quads"},
)
digest = hashes.Hash(hashes.SHA256())
digest.update(norm_form.encode("utf8"))
return digest.finalize().hex().encode("ascii")

9
core/tests/conftest.py Normal file
View File

@ -0,0 +1,9 @@
import pytest
from pyld import jsonld
from core.ld import builtin_document_loader
@pytest.fixture(scope="session", autouse=True)
def ldloader():
jsonld.set_document_loader(builtin_document_loader)

156
core/tests/test_signatures.py Normal file
View File

@ -0,0 +1,156 @@
import pytest
from asgiref.sync import async_to_sync
from django.test.client import RequestFactory
from pytest_httpx import HTTPXMock
from core.signatures import HttpSignature, LDSignature, VerificationError
# Our testing-only keypair
private_key = """-----BEGIN PRIVATE KEY-----
MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCzNJa9JIxQpOtQ
z8UQKXDPREF9DyBliGu3uPWo6DMnkOm7hoh2+nOryrWDqWOFaVK//n7kltHXUEbm
U3exh0/0iWfzx2AbNrI04csAvW/hRvHbHBnVTotSxzqTd3ESkpcSW4xVuz9aCcFR
kW3unSCO3fF0Lh8Jsy9N/CT6oTnwG+ZpeGvHVbh9xfR5Ww6zA7z8A6B17hbzdMd/
3qUPijyIb5se4cWVtGg/ZJ0X1syn9u9kpwUjhHlyWH/esMRHxPuW49BPZPhhKs1+
t//4xgZcRX515qFqPS2EtYgZAfh7M3TRv8uCSzL4TT+8ka9IUwKdV6TFaqH27bAG
KyJQfGaTAgMBAAECggEALZY5qFjlRtiFMfQApdlc5KTw4d7Yt2tqN3zaJUMYTD7d
boJNMbMJfNCetyT+d6Aw2D1ly0GglNzLhGkEQElzKfpQUt/Lj3CtCa3Mpd4K2Wxi
NwJhgfUulPqwaHYQchCPVLCsNNziw0VLA7Rymionb6B+/TaEV8PYy0ZSo90ir3UD
CL5t+IWgIPiy6pk1wGOmeB+tU4+V7/hFel+vPFNahafqVhLE311dfx2aOfweAEfN
e4JoPeJP1/fB+BVZMyVSAraKz6wheymBBNKKn/vpFsdd6it2AP4UZeFp6ma9wT9t
nk65IpHg1MBxazQd7621GrPH+ZnhMg62H/FEj6rIDQKBgQC1w1fEbk+zjI54DXU8
FAe5cJbZS89fMP5CtzlWKzTzfdaavT+5cUYp3XAv37tSGsqYAXxY+4bHGa+qdCQO
I41cmylWGNX2e29/p2BspDPM6YQ0Z21MxFRBTWvHFrhd0bF1cXKBKPttdkKvzOEP
6uNy+/QtRNn9xF/ZjaMHcyPPTQKBgQD8ZdOmZ3TMsYJchAjjseN8S+Objw2oZzmK
6I1ULJBz3DWiyCUfir+pMjSH4fsAf9zrHkiM7xUgMByTukVRt16BrT7TlEBanAxc
/AKdNB3f0pza829LCz1lMAUn+ngZLTmRR+1rQFXqTjhB+0peJzKiMli+9BBhL9Ry
jMeTuLHdXwKBgGiz9kL5KIBNX2RYnEfXYfu4l6zktrgnCNB1q1mv2fjJbG4GxkaU
sc47+Pwa7VUGid22PWMkwSa/7SlLbdmXMT8/QjiOZfJueHQYfrsWe6B2g+mMCrJG
BiL37jXpKJsiyA7XIxaz/OG5VgDfDGaW8B60dJv/JXPBQ1WW+Wq5MM+hAoGAAUdS
xykHAnJzwpw4n06rZFnOEV+sJgo/1GBRNvfy02NuMiDpbzt4tRa4BWgzqVD8gYRp
wa0EYmFcA7OR3lQbenSyOMgre0oHFgGA0eMNs7CRctqA2dR4vyZ7IDS4nwgHnqDK
pxxwUvuKdWsceVWhgAjZQj5iRtvDK8Fi0XDCFekCgYALTU1v5iMIpaRAe+eyA2B1
42qm4B/uhXznvOu2YXU6iJFmMgHGYgpa+Dq8uUjKtpn/LIFeX1KN0hH8z/0LW3gB
e7tN7taW0oLK3RQcEMfkZ7diE9x3LGqo/xMxsZMtxAr88p5eMEU/nxxznOqq+W9b
qxRbXYzEtHz+cW9+FZkyVw==
-----END PRIVATE KEY-----"""
public_key = """-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAszSWvSSMUKTrUM/FEClw
z0RBfQ8gZYhrt7j1qOgzJ5Dpu4aIdvpzq8q1g6ljhWlSv/5+5JbR11BG5lN3sYdP
9Iln88dgGzayNOHLAL1v4Ubx2xwZ1U6LUsc6k3dxEpKXEluMVbs/WgnBUZFt7p0g
jt3xdC4fCbMvTfwk+qE58BvmaXhrx1W4fcX0eVsOswO8/AOgde4W83THf96lD4o8
iG+bHuHFlbRoP2SdF9bMp/bvZKcFI4R5clh/3rDER8T7luPQT2T4YSrNfrf/+MYG
XEV+deahaj0thLWIGQH4ezN00b/Lgksy+E0/vJGvSFMCnVekxWqh9u2wBisiUHxm
kwIDAQAB
-----END PUBLIC KEY-----"""
public_key_id = "https://example.com/test-actor#test-key"
def test_sign_ld():
"""
Tests signing JSON-LD documents by round-tripping them through the
verifier.
"""
# Create the signature
document = {
"id": "https://example.com/test-create",
"type": "Create",
"actor": "https://example.com/test-actor",
"object": {
"id": "https://example.com/test-object",
"type": "Note",
},
}
signature_section = LDSignature.create_signature(
document,
private_key,
public_key_id,
)
# Check it and assign it to the document
assert "signatureValue" in signature_section
assert signature_section["type"] == "RsaSignature2017"
document["signature"] = signature_section
# Now verify it ourselves
LDSignature.verify_signature(document, public_key)
def test_verifying_ld():
"""
Tests verifying JSON-LD signatures from a known-good document
"""
document = {
"id": "https://example.com/test-create",
"type": "Create",
"actor": "https://example.com/test-actor",
"object": {"id": "https://example.com/test-object", "type": "Note"},
"signature": {
"@context": "https://w3id.org/identity/v1",
"creator": "https://example.com/test-actor#test-key",
"created": "2022-11-12T21:41:47Z",
"signatureValue": "nTHfkHqG4hegfnjpHucXtXDLDaIKi2Duk+NeCzqTtkjf4NneXsofbZY2tGew4uAooEe1UeM23PIyjWYnR16KwcD4YY8nMj8L3xY2czwQPScMM9n+KhSHzkWfX+iI4FWKbjpPI8M53EtTRJU+1qEjjmGUx03Ip0vfvT5821etIgvY4wLNhg3y7R8fevnNux+BeytcEV6gM4awJJ6RK0xrWGLyTgDNon5V5aNUjwcV/UVPy9UAQi1KYWtA74/F0Y4oPzL5CTudPpyiViyVHZQaal4r+ExzgSvGztqKxQeT1ya6gLXxbm1YQ+8UiGVSS8zoGhMFDEZWVsRPv7e0jm5wfA==",
"type": "RsaSignature2017",
},
}
# Ensure it verifies with correct data
LDSignature.verify_signature(document, public_key)
# Mutate it slightly and ensure it does not verify
with pytest.raises(VerificationError):
document["actor"] = "https://example.com/evil-actor"
LDSignature.verify_signature(document, public_key)
def test_sign_http(httpx_mock: HTTPXMock):
"""
Tests signing HTTP requests by round-tripping them through our verifier
"""
# Create document
document = {
"id": "https://example.com/test-create",
"type": "Create",
"actor": "https://example.com/test-actor",
"object": {
"id": "https://example.com/test-object",
"type": "Note",
},
}
# Send the signed request to the mock library
httpx_mock.add_response()
async_to_sync(HttpSignature.signed_request)(
uri="https://example.com/test-actor",
body=document,
private_key=private_key,
key_id=public_key_id,
)
# Retrieve it and construct a fake request object
outbound_request = httpx_mock.get_request()
fake_request = RequestFactory().post(
path="/test-actor",
data=outbound_request.content,
content_type=outbound_request.headers["content-type"],
HTTP_HOST="example.com",
HTTP_DATE=outbound_request.headers["date"],
HTTP_SIGNATURE=outbound_request.headers["signature"],
HTTP_DIGEST=outbound_request.headers["digest"],
)
# Verify that
HttpSignature.verify_request(fake_request, public_key)
def test_verify_http():
"""
Tests verifying HTTP requests against a known good example
"""
# Make our predictable request
fake_request = RequestFactory().post(
path="/test-actor",
data=b'{"id": "https://example.com/test-create", "type": "Create", "actor": "https://example.com/test-actor", "object": {"id": "https://example.com/test-object", "type": "Note"}}',
content_type="application/json",
HTTP_HOST="example.com",
HTTP_DATE="Sat, 12 Nov 2022 21:57:18 GMT",
HTTP_SIGNATURE='keyId="https://example.com/test-actor#test-key",headers="(request-target) host date digest content-type",signature="IRduYoDJIh90mprjUgOIdxY1iaBWHs5ou9vsDlcmSekg6DXMZTiXjmZxbNIrnpEbNFu3wTcqz1nv9H97Gp7orbYMuHm6j2ecxsvzSr37T9jxBbt3Ov3xSfuYWwhv6PuTWNxHtUQWNuAIc3wHDAQt8Flnak/uHe7swoAq4uHq2kt18iMW6CEV9XA5ESFho2HSUgRaifoNxJlIWbHYPJiP0t9aktgGBkpQoZ8ulOj3Ew4RwC1lwk9kzWiLIjU4tSAie8RbIy2g0aUvA1tQh9Uge1by3o7+349SL5iooj+B6WSCEvvjEl52wo3xoEQmv0ptYuSPLUgB9tP8q7DoHEc8Dw==",algorithm="rsa-sha256"',
HTTP_DIGEST="SHA-256=07sIbQ3GlOHWMbFMNajtPNtmUQXXu20UuvrIYLlI3kc=",
)
# Verify that
HttpSignature.verify_request(fake_request, public_key, skip_date=True)

2
requirements.txt
View File

@ -10,3 +10,5 @@ uvicorn~=0.19
gunicorn~=20.1.0
psycopg2~=2.9.5
bleach~=5.0.1
pytest-django~=4.5.2
pytest-httpx~=0.21

6
setup.cfg
View File

@ -7,6 +7,12 @@ max-line-length = 119
profile = black
multi_line_output = 3
[tool:pytest]
addopts = --tb=short
DJANGO_SETTINGS_MODULE = takahe.settings
filterwarnings =
ignore:There is no current event loop
[mypy]
warn_unused_ignores = True

6
users/admin.py
View File

@ -22,6 +22,12 @@ class UserEventAdmin(admin.ModelAdmin):
class IdentityAdmin(admin.ModelAdmin):
list_display = ["id", "handle", "actor_uri", "state", "local"]
raw_id_fields = ["users"]
actions = ["force_update"]
@admin.action(description="Force Update")
def force_update(self, request, queryset):
for instance in queryset:
instance.transition_perform("outdated")
@admin.register(Follow)

18
users/migrations/0002_identity_public_key_id.py Normal file
View File

@ -0,0 +1,18 @@
# Generated by Django 4.1.3 on 2022-11-12 21:29
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("users", "0001_initial"),
]
operations = [
migrations.AddField(
model_name="identity",
name="public_key_id",
field=models.TextField(blank=True, null=True),
),
]

9
users/models/follow.py
View File

@ -37,7 +37,8 @@ class FollowStates(StateGraph):
await HttpSignature.signed_request(
uri=follow.target.inbox_uri,
body=canonicalise(follow.to_ap()),
identity=follow.source,
private_key=follow.source.public_key,
key_id=follow.source.public_key_id,
)
return cls.local_requested
@ -56,7 +57,8 @@ class FollowStates(StateGraph):
await HttpSignature.signed_request(
uri=follow.source.inbox_uri,
body=canonicalise(follow.to_accept_ap()),
identity=follow.target,
private_key=follow.target.public_key,
key_id=follow.target.public_key_id,
)
return cls.accepted
@ -69,7 +71,8 @@ class FollowStates(StateGraph):
await HttpSignature.signed_request(
uri=follow.target.inbox_uri,
body=canonicalise(follow.to_undo_ap()),
identity=follow.source,
private_key=follow.source.public_key,
key_id=follow.source.public_key_id,
)
return cls.undone_remotely

36
users/models/identity.py
View File

@ -11,7 +11,6 @@ from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from django.db import models
from django.utils import timezone
from OpenSSL import crypto
from core.ld import canonicalise
from stator.models import State, StateField, StateGraph, StatorModel
@ -89,6 +88,7 @@ class Identity(StatorModel):
private_key = models.TextField(null=True, blank=True)
public_key = models.TextField(null=True, blank=True)
public_key_id = models.TextField(null=True, blank=True)
created = models.DateTimeField(auto_now_add=True)
updated = models.DateTimeField(auto_now=True)
@ -182,10 +182,6 @@ class Identity(StatorModel):
# TODO: Setting
return self.data_age > 60 * 24 * 24
@property
def key_id(self):
return self.actor_uri + "#main-key"
### Actor/Webfinger fetching ###
@classmethod
@ -242,6 +238,7 @@ class Identity(StatorModel):
"as:manuallyApprovesFollowers"
)
self.public_key = document.get("publicKey", {}).get("publicKeyPem")
self.public_key_id = document.get("publicKey", {}).get("id")
self.icon_uri = document.get("icon", {}).get("url")
self.image_uri = document.get("image", {}).get("url")
# Now go do webfinger with that info to see if we can get a canonical domain
@ -286,32 +283,3 @@ class Identity(StatorModel):
.decode("ascii")
)
self.save()
def sign(self, cleartext: str) -> bytes:
if not self.private_key:
raise ValueError("Cannot sign - no private key")
pkey = crypto.load_privatekey(
crypto.FILETYPE_PEM,
self.private_key.encode("ascii"),
)
return crypto.sign(
pkey,
cleartext.encode("ascii"),
"sha256",
)
def verify_signature(self, signature: bytes, cleartext: str) -> bool:
if not self.public_key:
raise ValueError("Cannot verify - no public key")
x509 = crypto.X509()
x509.set_pubkey(
crypto.load_publickey(
crypto.FILETYPE_PEM,
self.public_key.encode("ascii"),
)
)
try:
crypto.verify(x509, signature, cleartext.encode("ascii"), "sha256")
except crypto.Error:
return False
return True

71
users/views/identity.py
View File

@ -7,15 +7,18 @@ from django.conf import settings
from django.contrib.auth.decorators import login_required
from django.http import Http404, HttpResponse, HttpResponseBadRequest, JsonResponse
from django.shortcuts import redirect
from django.utils import timezone
from django.utils.decorators import method_decorator
from django.utils.http import parse_http_date
from django.views.decorators.csrf import csrf_exempt
from django.views.generic import FormView, TemplateView, View
from core.forms import FormHelper
from core.ld import canonicalise
from core.signatures import HttpSignature
from core.signatures import (
HttpSignature,
LDSignature,
VerificationError,
VerificationFormatError,
)
from users.decorators import identity_required
from users.models import Domain, Follow, Identity, IdentityStates, InboxMessage
from users.shortcuts import by_handle_or_404
@ -167,7 +170,7 @@ class Actor(View):
"inbox": identity.actor_uri + "inbox/",
"preferredUsername": identity.username,
"publicKey": {
"id": identity.key_id,
"id": identity.public_key_id,
"owner": identity.actor_uri,
"publicKeyPem": identity.public_key,
},
@ -188,37 +191,8 @@ class Inbox(View):
"""
def post(self, request, handle):
# Verify body digest
if "HTTP_DIGEST" in request.META:
expected_digest = HttpSignature.calculate_digest(request.body)
if request.META["HTTP_DIGEST"] != expected_digest:
print("Wrong digest")
return HttpResponseBadRequest("Digest is incorrect")
# Verify date header
if "HTTP_DATE" in request.META:
header_date = parse_http_date(request.META["HTTP_DATE"])
if abs(timezone.now().timestamp() - header_date) > 60:
print(
f"Date mismatch - they sent {header_date}, now is {timezone.now().timestamp()}"
)
return HttpResponseBadRequest("Date is too far away")
# Get the signature details
if "HTTP_SIGNATURE" not in request.META:
print("No signature")
return HttpResponseBadRequest("No signature present")
signature_details = HttpSignature.parse_signature(
request.META["HTTP_SIGNATURE"]
)
# Reject unknown algorithms
if signature_details["algorithm"] != "rsa-sha256":
print("Unknown sig algo")
return HttpResponseBadRequest("Unknown signature algorithm")
# Create the signature payload
headers_string = HttpSignature.headers_from_request(
request, signature_details["headers"]
)
# Load the LD
document = canonicalise(json.loads(request.body))
document = canonicalise(json.loads(request.body), include_security=True)
# Find the Identity by the actor on the incoming item
# This ensures that the signature used for the headers matches the actor
# described in the payload.
@ -229,12 +203,29 @@ class Inbox(View):
if not identity.public_key:
print("Cannot get actor")
return HttpResponseBadRequest("Cannot retrieve actor")
if not identity.verify_signature(
signature_details["signature"], headers_string
):
print("Bad signature!")
print(document)
return HttpResponseUnauthorized("Bad signature")
# If there's a "signature" payload, verify against that
if "signature" in document:
try:
LDSignature.verify_signature(document, identity.public_key)
except VerificationFormatError as e:
print("Bad LD signature format:", e.args[0])
return HttpResponseBadRequest(e.args[0])
except VerificationError:
print("Bad LD signature")
return HttpResponseUnauthorized("Bad signature")
# Otherwise, verify against the header (assuming it's the same actor)
else:
try:
HttpSignature.verify_request(
request,
identity.public_key,
)
except VerificationFormatError as e:
print("Bad HTTP signature format:", e.args[0])
return HttpResponseBadRequest(e.args[0])
except VerificationError:
print("Bad HTTP signature")
return HttpResponseUnauthorized("Bad signature")
# Hand off the item to be processed by the queue
InboxMessage.objects.create(message=document)
return HttpResponse(status=202)