fixed reverse chronological ordering in a pair of FTs clogging the pipeline; added ActivityPub to project; new apps.ap for WebFinger, Actor, Outbox views; apps.lyric.models now contains ap_public_key, ap_private_key fields + ensure_keypair(); new apps.lyric migration accordingly; new in drama.models are to_activity() w. JoinGate, SelectRole, Create compat. & None verb support; new core.urls for /.well-known/webfinger + /ap/ included; cryptography installed, added to reqs.txt; 24 new green UTs & ITs; in sum, project is now read-only ActivityPub node
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
This commit is contained in:
0
src/apps/ap/__init__.py
Normal file
0
src/apps/ap/__init__.py
Normal file
7
src/apps/ap/apps.py
Normal file
7
src/apps/ap/apps.py
Normal file
@@ -0,0 +1,7 @@
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class ApConfig(AppConfig):
|
||||
default_auto_field = "django.db.models.BigAutoField"
|
||||
name = "apps.ap"
|
||||
label = "ap"
|
||||
0
src/apps/ap/tests/__init__.py
Normal file
0
src/apps/ap/tests/__init__.py
Normal file
0
src/apps/ap/tests/integrated/__init__.py
Normal file
0
src/apps/ap/tests/integrated/__init__.py
Normal file
119
src/apps/ap/tests/integrated/test_ap_views.py
Normal file
119
src/apps/ap/tests/integrated/test_ap_views.py
Normal file
@@ -0,0 +1,119 @@
|
||||
import json
|
||||
|
||||
from django.test import TestCase
|
||||
|
||||
from apps.drama.models import GameEvent, record
|
||||
from apps.epic.models import Room
|
||||
from apps.lyric.models import User
|
||||
|
||||
|
||||
class WebFingerTest(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="actor@test.io", username="actor")
|
||||
|
||||
def test_returns_jrd_for_known_user(self):
|
||||
response = self.client.get(
|
||||
"/.well-known/webfinger",
|
||||
{"resource": "acct:actor@earthmanrpg.me"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response["Content-Type"], "application/jrd+json")
|
||||
|
||||
def test_jrd_links_to_actor_url(self):
|
||||
response = self.client.get(
|
||||
"/.well-known/webfinger",
|
||||
{"resource": "acct:actor@earthmanrpg.me"},
|
||||
)
|
||||
data = json.loads(response.content)
|
||||
hrefs = [link["href"] for link in data["links"]]
|
||||
self.assertTrue(any("/ap/users/actor/" in href for href in hrefs))
|
||||
|
||||
def test_returns_404_for_unknown_user(self):
|
||||
response = self.client.get(
|
||||
"/.well-known/webfinger",
|
||||
{"resource": "acct:nobody@earthmanrpg.me"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 404)
|
||||
|
||||
def test_returns_400_for_missing_resource(self):
|
||||
response = self.client.get("/.well-known/webfinger")
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
|
||||
class ActorViewTest(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="actor@test.io", username="actor")
|
||||
|
||||
def test_returns_200_for_known_user(self):
|
||||
response = self.client.get("/ap/users/actor/")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_returns_activity_json_content_type(self):
|
||||
response = self.client.get("/ap/users/actor/")
|
||||
self.assertEqual(response["Content-Type"], "application/activity+json")
|
||||
|
||||
def test_actor_has_required_fields(self):
|
||||
response = self.client.get("/ap/users/actor/")
|
||||
data = json.loads(response.content)
|
||||
self.assertEqual(data["type"], "Person")
|
||||
self.assertIn("id", data)
|
||||
self.assertIn("outbox", data)
|
||||
self.assertIn("publicKey", data)
|
||||
|
||||
def test_requires_no_authentication(self):
|
||||
# AP Actor endpoints must be publicly accessible
|
||||
self.client.logout()
|
||||
response = self.client.get("/ap/users/actor/")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_returns_404_for_unknown_user(self):
|
||||
response = self.client.get("/ap/users/nobody/")
|
||||
self.assertEqual(response.status_code, 404)
|
||||
|
||||
|
||||
class OutboxViewTest(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="actor@test.io", username="actor")
|
||||
self.room = Room.objects.create(name="Test Room", owner=self.user)
|
||||
record(
|
||||
self.room, GameEvent.SLOT_FILLED, actor=self.user,
|
||||
slot_number=1, token_type="coin",
|
||||
token_display="Coin", renewal_days=7,
|
||||
)
|
||||
record(
|
||||
self.room, GameEvent.ROLE_SELECTED, actor=self.user,
|
||||
role="PC", slot_number=1, role_display="Player",
|
||||
)
|
||||
# INVITE_SENT is unsupported — should be excluded from outbox
|
||||
record(self.room, GameEvent.INVITE_SENT, actor=self.user)
|
||||
|
||||
def test_returns_200(self):
|
||||
response = self.client.get("/ap/users/actor/outbox/")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_returns_activity_json_content_type(self):
|
||||
response = self.client.get("/ap/users/actor/outbox/")
|
||||
self.assertEqual(response["Content-Type"], "application/activity+json")
|
||||
|
||||
def test_outbox_is_ordered_collection(self):
|
||||
response = self.client.get("/ap/users/actor/outbox/")
|
||||
data = json.loads(response.content)
|
||||
self.assertEqual(data["type"], "OrderedCollection")
|
||||
|
||||
def test_total_items_excludes_unsupported_verbs(self):
|
||||
response = self.client.get("/ap/users/actor/outbox/")
|
||||
data = json.loads(response.content)
|
||||
# 2 supported events (SLOT_FILLED + ROLE_SELECTED); INVITE_SENT excluded
|
||||
self.assertEqual(data["totalItems"], 2)
|
||||
|
||||
def test_requires_no_authentication(self):
|
||||
self.client.logout()
|
||||
response = self.client.get("/ap/users/actor/outbox/")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_returns_404_for_unknown_user(self):
|
||||
response = self.client.get("/ap/users/nobody/outbox/")
|
||||
self.assertEqual(response.status_code, 404)
|
||||
0
src/apps/ap/tests/unit/__init__.py
Normal file
0
src/apps/ap/tests/unit/__init__.py
Normal file
88
src/apps/ap/tests/unit/test_activity.py
Normal file
88
src/apps/ap/tests/unit/test_activity.py
Normal file
@@ -0,0 +1,88 @@
|
||||
from django.test import TestCase
|
||||
|
||||
from apps.drama.models import GameEvent, record
|
||||
from apps.epic.models import Room
|
||||
from apps.lyric.models import User
|
||||
|
||||
|
||||
BASE = "https://earthmanrpg.me"
|
||||
|
||||
|
||||
class ToActivityTest(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="actor@test.io", username="testactor")
|
||||
self.room = Room.objects.create(name="Test Room", owner=self.user)
|
||||
|
||||
def _record(self, verb, **data):
|
||||
return record(self.room, verb, actor=self.user, **data)
|
||||
|
||||
def test_slot_filled_returns_join_gate_activity(self):
|
||||
event = self._record(
|
||||
GameEvent.SLOT_FILLED,
|
||||
slot_number=1, token_type="coin",
|
||||
token_display="Coin", renewal_days=7,
|
||||
)
|
||||
activity = event.to_activity(BASE)
|
||||
self.assertIsNotNone(activity)
|
||||
self.assertEqual(activity["type"], "earthman:JoinGate")
|
||||
|
||||
def test_role_selected_returns_select_role_activity(self):
|
||||
event = self._record(
|
||||
GameEvent.ROLE_SELECTED,
|
||||
role="PC", slot_number=1, role_display="Player",
|
||||
)
|
||||
activity = event.to_activity(BASE)
|
||||
self.assertIsNotNone(activity)
|
||||
self.assertEqual(activity["type"], "earthman:SelectRole")
|
||||
|
||||
def test_room_created_returns_create_activity(self):
|
||||
event = self._record(GameEvent.ROOM_CREATED)
|
||||
activity = event.to_activity(BASE)
|
||||
self.assertIsNotNone(activity)
|
||||
self.assertEqual(activity["type"], "Create")
|
||||
|
||||
def test_unsupported_verb_returns_none(self):
|
||||
event = self._record(GameEvent.INVITE_SENT)
|
||||
self.assertIsNone(event.to_activity(BASE))
|
||||
|
||||
def test_activity_contains_actor_url(self):
|
||||
event = self._record(
|
||||
GameEvent.ROLE_SELECTED,
|
||||
role="PC", slot_number=1, role_display="Player",
|
||||
)
|
||||
activity = event.to_activity(BASE)
|
||||
self.assertIn(BASE, activity["actor"])
|
||||
|
||||
def test_activity_contains_object_url(self):
|
||||
event = self._record(
|
||||
GameEvent.SLOT_FILLED,
|
||||
slot_number=1, token_type="coin",
|
||||
token_display="Coin", renewal_days=7,
|
||||
)
|
||||
activity = event.to_activity(BASE)
|
||||
self.assertIn(str(self.room.id), activity["object"])
|
||||
|
||||
|
||||
class EnsureKeypairTest(TestCase):
|
||||
|
||||
def test_ensure_keypair_populates_both_fields(self):
|
||||
user = User.objects.create(email="keys@test.io")
|
||||
self.assertEqual(user.ap_public_key, "")
|
||||
self.assertEqual(user.ap_private_key, "")
|
||||
user.ensure_keypair()
|
||||
self.assertTrue(user.ap_public_key.startswith("-----BEGIN PUBLIC KEY-----"))
|
||||
self.assertTrue(user.ap_private_key.startswith("-----BEGIN PRIVATE KEY-----"))
|
||||
|
||||
def test_ensure_keypair_persists_to_db(self):
|
||||
user = User.objects.create(email="persist@test.io")
|
||||
user.ensure_keypair()
|
||||
refreshed = User.objects.get(pk=user.pk)
|
||||
self.assertTrue(refreshed.ap_public_key.startswith("-----BEGIN PUBLIC KEY-----"))
|
||||
|
||||
def test_ensure_keypair_is_idempotent(self):
|
||||
user = User.objects.create(email="idem@test.io")
|
||||
user.ensure_keypair()
|
||||
original_pub = user.ap_public_key
|
||||
user.ensure_keypair()
|
||||
self.assertEqual(user.ap_public_key, original_pub)
|
||||
10
src/apps/ap/urls.py
Normal file
10
src/apps/ap/urls.py
Normal file
@@ -0,0 +1,10 @@
|
||||
from django.urls import path
|
||||
|
||||
from . import views
|
||||
|
||||
app_name = "ap"
|
||||
|
||||
urlpatterns = [
|
||||
path("users/<str:username>/", views.actor, name="actor"),
|
||||
path("users/<str:username>/outbox/", views.outbox, name="outbox"),
|
||||
]
|
||||
83
src/apps/ap/views.py
Normal file
83
src/apps/ap/views.py
Normal file
@@ -0,0 +1,83 @@
|
||||
import json
|
||||
|
||||
from django.http import HttpResponse, JsonResponse
|
||||
from django.shortcuts import get_object_or_404
|
||||
|
||||
from apps.lyric.models import User
|
||||
|
||||
|
||||
AP_CONTEXT = [
|
||||
"https://www.w3.org/ns/activitystreams",
|
||||
{"earthman": "https://earthmanrpg.me/ns#"},
|
||||
]
|
||||
|
||||
|
||||
def _base_url(request):
|
||||
return f"{request.scheme}://{request.get_host()}"
|
||||
|
||||
|
||||
def _ap_response(data):
|
||||
return HttpResponse(
|
||||
json.dumps(data),
|
||||
content_type="application/activity+json",
|
||||
)
|
||||
|
||||
|
||||
def webfinger(request):
|
||||
resource = request.GET.get("resource", "")
|
||||
if not resource:
|
||||
return HttpResponse(status=400)
|
||||
# Expect acct:username@host
|
||||
if not resource.startswith("acct:"):
|
||||
return HttpResponse(status=400)
|
||||
username = resource[len("acct:"):].split("@")[0]
|
||||
user = get_object_or_404(User, username=username)
|
||||
base = _base_url(request)
|
||||
data = {
|
||||
"subject": resource,
|
||||
"links": [
|
||||
{
|
||||
"rel": "self",
|
||||
"type": "application/activity+json",
|
||||
"href": f"{base}/ap/users/{user.username}/",
|
||||
}
|
||||
],
|
||||
}
|
||||
return HttpResponse(json.dumps(data), content_type="application/jrd+json")
|
||||
|
||||
|
||||
def actor(request, username):
|
||||
user = get_object_or_404(User, username=username)
|
||||
user.ensure_keypair()
|
||||
base = _base_url(request)
|
||||
actor_url = f"{base}/ap/users/{username}/"
|
||||
data = {
|
||||
"@context": AP_CONTEXT,
|
||||
"id": actor_url,
|
||||
"type": "Person",
|
||||
"preferredUsername": username,
|
||||
"inbox": f"{actor_url}inbox/",
|
||||
"outbox": f"{actor_url}outbox/",
|
||||
"publicKey": {
|
||||
"id": f"{actor_url}#main-key",
|
||||
"owner": actor_url,
|
||||
"publicKeyPem": user.ap_public_key,
|
||||
},
|
||||
}
|
||||
return _ap_response(data)
|
||||
|
||||
|
||||
def outbox(request, username):
|
||||
user = get_object_or_404(User, username=username)
|
||||
base = _base_url(request)
|
||||
events = user.game_events.select_related("room").order_by("timestamp")
|
||||
activities = [a for e in events if (a := e.to_activity(base)) is not None]
|
||||
actor_url = f"{base}/ap/users/{username}/"
|
||||
data = {
|
||||
"@context": AP_CONTEXT,
|
||||
"id": f"{actor_url}outbox/",
|
||||
"type": "OrderedCollection",
|
||||
"totalItems": len(activities),
|
||||
"orderedItems": activities,
|
||||
}
|
||||
return _ap_response(data)
|
||||
@@ -78,6 +78,35 @@ class GameEvent(models.Model):
|
||||
return "All roles assigned"
|
||||
return self.verb
|
||||
|
||||
def to_activity(self, base_url):
|
||||
"""Serialise this event as an AS2 Activity dict, or None if unsupported."""
|
||||
if not self.actor or not self.actor.username:
|
||||
return None
|
||||
actor_url = f"{base_url}/ap/users/{self.actor.username}/"
|
||||
room_url = f"{base_url}/gameboard/room/{self.room_id}/"
|
||||
if self.verb == self.SLOT_FILLED:
|
||||
return {
|
||||
"type": "earthman:JoinGate",
|
||||
"actor": actor_url,
|
||||
"object": room_url,
|
||||
"summary": self.to_prose(),
|
||||
}
|
||||
if self.verb == self.ROLE_SELECTED:
|
||||
return {
|
||||
"type": "earthman:SelectRole",
|
||||
"actor": actor_url,
|
||||
"object": room_url,
|
||||
"summary": self.to_prose(),
|
||||
}
|
||||
if self.verb == self.ROOM_CREATED:
|
||||
return {
|
||||
"type": "Create",
|
||||
"actor": actor_url,
|
||||
"object": room_url,
|
||||
"summary": self.to_prose(),
|
||||
}
|
||||
return None
|
||||
|
||||
def __str__(self):
|
||||
actor = self.actor.email if self.actor else "system"
|
||||
return f"[{self.timestamp:%Y-%m-%d %H:%M}] {actor} → {self.verb}"
|
||||
|
||||
23
src/apps/lyric/migrations/0017_ap_keypair_fields.py
Normal file
23
src/apps/lyric/migrations/0017_ap_keypair_fields.py
Normal file
@@ -0,0 +1,23 @@
|
||||
# Generated by Django 6.0 on 2026-04-02 19:12
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('lyric', '0016_backfill_unlocked_decks'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='user',
|
||||
name='ap_private_key',
|
||||
field=models.TextField(blank=True, default=''),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='user',
|
||||
name='ap_public_key',
|
||||
field=models.TextField(blank=True, default=''),
|
||||
),
|
||||
]
|
||||
@@ -44,6 +44,8 @@ class User(AbstractBaseUser):
|
||||
unlocked_decks = models.ManyToManyField(
|
||||
"epic.DeckVariant", blank=True, related_name="unlocked_by",
|
||||
)
|
||||
ap_public_key = models.TextField(blank=True, default="")
|
||||
ap_private_key = models.TextField(blank=True, default="")
|
||||
|
||||
is_staff = models.BooleanField(default=False)
|
||||
is_superuser = models.BooleanField(default=False)
|
||||
@@ -52,6 +54,24 @@ class User(AbstractBaseUser):
|
||||
REQUIRED_FIELDS = []
|
||||
USERNAME_FIELD = "email"
|
||||
|
||||
def ensure_keypair(self):
|
||||
"""Generate and persist an RSA-2048 keypair if not already set."""
|
||||
if self.ap_public_key:
|
||||
return
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
self.ap_public_key = private_key.public_key().public_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PublicFormat.SubjectPublicKeyInfo,
|
||||
).decode()
|
||||
self.ap_private_key = private_key.private_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PrivateFormat.PKCS8,
|
||||
serialization.NoEncryption(),
|
||||
).decode()
|
||||
self.save(update_fields=["ap_public_key", "ap_private_key"])
|
||||
|
||||
def has_perm(self, perm, obj=None):
|
||||
return self.is_superuser
|
||||
|
||||
|
||||
@@ -62,6 +62,7 @@ INSTALLED_APPS = [
|
||||
'apps.epic',
|
||||
'apps.drama',
|
||||
'apps.billboard',
|
||||
'apps.ap',
|
||||
# Custom apps
|
||||
'apps.api',
|
||||
'apps.applets',
|
||||
|
||||
@@ -2,6 +2,7 @@ from django.contrib import admin
|
||||
from django.http import HttpResponse
|
||||
from django.urls import include, path
|
||||
|
||||
from apps.ap import views as ap_views
|
||||
from apps.dashboard import views as dash_views
|
||||
|
||||
|
||||
@@ -14,6 +15,8 @@ urlpatterns = [
|
||||
path('gameboard/', include('apps.gameboard.urls')),
|
||||
path('gameboard/', include('apps.epic.urls')),
|
||||
path('billboard/', include('apps.billboard.urls')),
|
||||
path('ap/', include('apps.ap.urls')),
|
||||
path('.well-known/webfinger', ap_views.webfinger, name='webfinger'),
|
||||
]
|
||||
|
||||
# Please remove the following urlpattern
|
||||
|
||||
@@ -341,7 +341,8 @@ class BillscrollEntryLayoutTest(FunctionalTest):
|
||||
|
||||
def test_recent_event_shows_time_format(self):
|
||||
events = self._go_to_scroll()
|
||||
recent_ts = events[0].find_element(By.CSS_SELECTOR, ".drama-event-time")
|
||||
# events[0] is the backdated record (oldest); events[1] is fresh
|
||||
recent_ts = events[1].find_element(By.CSS_SELECTOR, ".drama-event-time")
|
||||
self.assertRegex(recent_ts.text, r"\d+:\d+\s+[ap]\.m\.")
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
@@ -350,5 +351,6 @@ class BillscrollEntryLayoutTest(FunctionalTest):
|
||||
|
||||
def test_old_event_shows_date_with_year(self):
|
||||
events = self._go_to_scroll()
|
||||
old_ts = events[1].find_element(By.CSS_SELECTOR, ".drama-event-time")
|
||||
# events[0] is the backdated record (oldest first, ascending order)
|
||||
old_ts = events[0].find_element(By.CSS_SELECTOR, ".drama-event-time")
|
||||
self.assertRegex(old_ts.text, r"\d{2}\s+[A-Z][a-z]{2}\s+\d{4}")
|
||||
|
||||
Reference in New Issue
Block a user