Mastodon isn't notify the Mention?

[AmaseCocoa]

This issue is forked from #11. I checked notify mention from Misskey.

I will test in own mastodon instance for check log.

[jze]

https://activitypub.academy is a great way to see what is going on behind the scenes of a Mastodon server.

[AmaseCocoa]

This is likely not suitable for debugging this issue. Since the problem stems maybe from compatibility with Mastodon itself, we need to examine more detailed logs, and in that case, self-hosting is the appropriate approach.

[AmaseCocoa]

The scary thing is that when I tested it with the following code, the notification was successful (however, Misskey has an issue where the job is not created properly and therefore cannot be accepted in the first place)

import asyncio
import json
import logging
import os
import uuid
from datetime import datetime

from cryptography.hazmat.primitives import serialization as crypto_serialization
from cryptography.hazmat.primitives.asymmetric import ed25519, rsa

from apkit.client.asyncio import ActivityPubClient
from apkit.models import Create, CryptographicKey, Note, Person
from apkit.types import ActorKey

HOST = ""
USER_ID = "demo"
TARGET_ID = "https://activitypub.academy/users/"

# --- Logging Setup ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# --- Key Persistence ---
KEY_FILE = "private_key.pem"

if os.path.exists(KEY_FILE):
    logger.info(f"Loading existing private key from {KEY_FILE}.")
    with open(KEY_FILE, "rb") as f:
        private_key = crypto_serialization.load_pem_private_key(
            f.read(), password=None
        )
else:
    logger.info(
        f"No key file found. Generating new private key and saving to {KEY_FILE}."
    )
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    with open(KEY_FILE, "wb") as f:
        _ = f.write(
            private_key.private_bytes(
                encoding=crypto_serialization.Encoding.PEM,
                format=crypto_serialization.PrivateFormat.PKCS8,
                encryption_algorithm=crypto_serialization.NoEncryption(),
            )
        )

# --- Key Persistence ---
KEY_FILE = "private_key_ed25519.pem"

if os.path.exists(KEY_FILE):
    logger.info(f"Loading existing private key from {KEY_FILE}.")
    with open(KEY_FILE, "rb") as f:
        private_key_ed25519 = crypto_serialization.load_pem_private_key(
            f.read(), password=None
        )
else:
    logger.info(
        f"No key file found. Generating new private key and saving to {KEY_FILE}."
    )
    private_key_ed25519 = ed25519.Ed25519PrivateKey.generate()
    with open(KEY_FILE, "wb") as f:
        f.write(
            private_key_ed25519.private_bytes(
                encoding=crypto_serialization.Encoding.PEM,
                format=crypto_serialization.PrivateFormat.PKCS8,
                encryption_algorithm=crypto_serialization.NoEncryption(),
            )
        )

public_key_pem = (
    private_key.public_key()
    .public_bytes(
        encoding=crypto_serialization.Encoding.PEM,
        format=crypto_serialization.PublicFormat.SubjectPublicKeyInfo,
    )
    .decode("utf-8")
)


async def main():
    async with ActivityPubClient() as client:
        # Fetch a remote Actor
        target_actor = await client.actor.fetch(TARGET_ID)
        print(f"Fetched actor: {target_actor.name}")

        # Get the inbox URL from the actor's profile
        inbox_url = target_actor.inbox
        if not inbox_url:
            raise Exception("Could not find actor's inbox URL")

        logger.info(f"Found actor's inbox: {inbox_url}")

        # Create actor
        actor = Person(
            id=f"https://{HOST}/users/{USER_ID}",
            name="apkit Demo",
            preferredUsername="demo",
            summary="This is a demo actor powered by apkit!",
            inbox=f"https://{HOST}/users/{USER_ID}/inbox",
            outbox=f"https://{HOST}/users/{USER_ID}/outbox",
            publicKey=CryptographicKey(
                id=f"https://{HOST}/users/{USER_ID}#main-key",
                owner=f"https://{HOST}/users/{USER_ID}",
                publicKeyPem=public_key_pem,
            ),
        )

        # Create note
        t = f'<p><span class="h-card" translate="no"><a href="{target_actor.url}" class="u-url mention">@<span>{target_actor.preferredUsername}</span></a></span></p>'
        note_id = str(uuid.uuid4())
        note = Note(
            id=f"https://{HOST}/notes/{note_id}",
            attributedTo=actor.id,
            content=f"<p>{t} Hello from apkit</p>",
            published=datetime.utcnow().isoformat() + "Z",
            to=["https://www.w3.org/ns/activitystreams#Public", target_actor.id], # target_actor.id
            #cc=["https://www.w3.org/ns/activitystreams#Public"],
        )
        with open(f"./data/notes/{note_id}", "w") as f:
            json.dump(note.to_json(), f)

        # Create activity
        create_id = str(uuid.uuid4())
        create = Create(
            id=f"https://{HOST}/creates/{create_id}",
            actor=actor.id,
            object=note,
            published=datetime.utcnow().isoformat() + "Z",
            to=note.to,
            cc=note.cc,
        )
        with open(f"./data/creates/{create_id}", "w") as f:
            json.dump(create.to_json(), f)

        #print(note.to_json())

        # Deliver the activity
        logger.info("Delivering activity...")

        #    key_id=actor.publicKey.id,
        #    signature=private_key,
        resp = await client.post(
            inbox_url,
            signatures=[
                ActorKey(
                    key_id=f"https://{HOST}/users/demo#main-key",
                    private_key=private_key
                )
            ],
            sign_with=['draft-cavage'], # , 'rsa2017', 'fep8b32'
            json=create.to_json(),
        )
        #logger.info(f"Delivery result: {await resp.json()}")
        logger.info(f"Delivery result: {resp.status}")


if __name__ == "__main__":
    asyncio.run(main())