Generate test messages for validation, instead of supplying literals

django_kepi/validation.py

@@ -5,6 +5,7 @@ import uuid
 from django.conf import settings
 from urllib.parse import urlparse
 from django_kepi import find
+from httpsig.verify import HeaderVerifier
 
 logger = logging.getLogger(name='django_kepi')
 

requirements.txt

@@ -6,4 +6,5 @@ requests-http-signature
 cryptography
 django-celery
 httpretty
+httpsig
 django-celery-results

tests/test_validation.py

@@ -1,21 +1,77 @@
+import json
+import httpsig
 from django.test import TestCase
 from django.db.models.query import QuerySet
 from django_kepi.models import IncomingMessage, validate
 from unittest.mock import patch
 import django_kepi.validation
 
-def _test_message():
-    result = IncomingMessage(
-            content_type = "application/activity+json",
-            date = "Thu, 04 Apr 2019 21:12:11 GMT",
-            digest = "SHA-256=MJRFYant/jJdSotCYRY4n1PtDFVIvYJxfCxydrimx/o=",
-            host = "marnanel.org",
-            signature = "keyId=\"https://queer.party/users/marnanel#main-key\",algorithm=\"rsa-sha256\",headers=\"(request-target) host date digest content-type\",signature=\"Db3Ua+VMRkfMc05p0r2te9pFHCOjAbtifagR7Q+Hjhc+VPsf5QNkPzMsIbSKuhZLRLkj3xDzbsiNrLg8zZOcM0MQCQGeL8ACgpBOm/nqB+USdUOcC7pdy1X/zN+oKbVroho3FgC53xHoEz3l7LVSTifzmhyf2/P8yFO7UpbcisX1fgazrjMhBPOI2Mv+JkDgBpBinzRS5V+O+RaJ9Hw8NjgI+1zvE/BHMLVttJYaI2vGt5ParqpVq3DLh5NeHuyXq6wRLI2ee5TxYAZiETpz3XKFpoOZOBi2P98mjTUAbFKyFc5C5/UfSgksy6fyuWtbmBouPXsKf1lEOmdC+dUt0Q==\"",
-            body = "{\"@context\":[\"https://www.w3.org/ns/activitystreams\",\"https://w3id.org/security/v1\",{\"manuallyApprovesFollowers\":\"as:manuallyApprovesFollowers\",\"sensitive\":\"as:sensitive\",\"movedTo\":{\"@id\":\"as:movedTo\",\"@type\":\"@id\"},\"alsoKnownAs\":{\"@id\":\"as:alsoKnownAs\",\"@type\":\"@id\"},\"Hashtag\":\"as:Hashtag\",\"ostatus\":\"http://ostatus.org#\",\"atomUri\":\"ostatus:atomUri\",\"inReplyToAtomUri\":\"ostatus:inReplyToAtomUri\",\"conversation\":\"ostatus:conversation\",\"toot\":\"http://joinmastodon.org/ns#\",\"Emoji\":\"toot:Emoji\",\"focalPoint\":{\"@container\":\"@list\",\"@id\":\"toot:focalPoint\"},\"featured\":{\"@id\":\"toot:featured\",\"@type\":\"@id\"},\"schema\":\"http://schema.org#\",\"PropertyValue\":\"schema:PropertyValue\",\"value\":\"schema:value\"}],\"id\":\"https://queer.party/04b065f8-81c4-408e-bec3-9fb1f7c06408\",\"type\":\"Follow\",\"actor\":\"https://queer.party/users/marnanel\",\"object\":\"https://marnanel.org/users/marnanel\"}",
-            actor = "https://queer.party/users/marnanel",
-            key_id = "https://queer.party/users/marnanel#main-key",
-                )
-    return result
+MESSAGE_CONTEXT = ["https://www.w3.org/ns/activitystreams",
+        "https://w3id.org/security/v1",
+        {"manuallyApprovesFollowers":"as:manuallyApprovesFollowers",
+            "sensitive":"as:sensitive",
+            "movedTo":{"@id":"as:movedTo",
+                "@type":"@id"},
+            "alsoKnownAs":{"@id":"as:alsoKnownAs",
+                "@type":"@id"},
+            "Hashtag":"as:Hashtag",
+            "ostatus":"http://ostatus.org#",
+            "atomUri":"ostatus:atomUri",
+            "inReplyToAtomUri":"ostatus:inReplyToAtomUri",
+            "conversation":"ostatus:conversation",
+            "toot":"http://joinmastodon.org/ns#",
+            "Emoji":"toot:Emoji",
+            "focalPoint":{"@container":"@list",
+                "@id":"toot:focalPoint"},
+            "featured":{"@id":"toot:featured",
+                "@type":"@id"},
+            "schema":"http://schema.org#",
+            "PropertyValue":"schema:PropertyValue",
+            "value":"schema:value"}]
+
+def _test_message(secret='', **fields):
+
+    body = dict([(f[2:],v) for f,v in fields.items() if f.startswith('f_')])
+    body['@context'] = MESSAGE_CONTEXT
+
+    headers = {
+            'content-type': "application/activity+json",
+            'date': "Thu, 04 Apr 2019 21:12:11 GMT",
+            'host': "europa.example.org",
+            }
+
+    if 'key_id' in fields:
+        key_id = fields['key_id']
+    else:
+        key_id = body['actor']+'#main-key'
+
+    signer = httpsig.HeaderSigner(
+            secret=secret,
+            algorithm='rsa-sha256',
+            key_id = key_id,
+            headers=['(request-target)', 'host', 'date', 'content-type'],
+            )
+
+    headers = signer.sign(
+            headers,
+            method='POST',
+            path='/inbox',
+            )
+
+    SIGNATURE = 'Signature'
+    if headers['Authorization'].startswith(SIGNATURE):
+        headers['Signature'] = headers['Authorization'][len(SIGNATURE)+1:]
+
+    return IncomingMessage(
+            content_type = headers['content-type'],
+            date = headers['date'],
+            digest = '', # FIXME ???
+            host = headers['host'],
+            signature = headers['Signature'],
+            body = json.dumps(body, sort_keys=True),
+            actor = body['actor'],
+            key_id = key_id,
+            )
 
 class TestValidation(TestCase):
 
@@ -24,9 +80,17 @@ class TestValidation(TestCase):
     def test_local_lookup(self, mock_key_get, mock_fetch):
         
         mock_key_get.return_value = None
+        keys = json.load(open('tests/keys/keys-0000.json', 'r'))
 
-        message = _test_message()
+        FRED = "https://remote.example.com/users/fred"
+        message = _test_message(
+                f_id="https://queer.party/04b065f8-81c4-408e-bec3-9fb1f7c06408",
+                f_type="Follow",
+                f_actor=FRED,
+                f_object="https://local.example.org/users/alice",
+                secret = keys['private'],
+                )
         validate(message)
-        mock_fetch.assert_called_once_with('https://queer.party/users/marnanel')
-        mock_key_get.assert_called_once_with(owner='https://queer.party/users/marnanel')
+        mock_fetch.assert_called_once_with(FRED)
+        mock_key_get.assert_called_once_with(owner=FRED)