Mobile Push Notifications skill for Claude Code

#!/usr/bin/env python3
"""Validate an APNs payload or an FCM HTTP v1 request body (stdlib only).

Usage:
  python3 check_push_payload.py FILE.json [--type apns|fcm] [--push-type alert|background|voip|liveactivity]

Type is auto-detected: {"message": {...}} -> FCM v1, {"aps": {...}} -> APNs.
Errors exit 1; warnings are printed but do not fail. Exit 2 on unreadable input.
"""
import argparse
import json
import sys

APNS_LIMIT = 4096
APNS_VOIP_LIMIT = 5120
FCM_LIMIT = 4096
INTERRUPTION_LEVELS = {"passive", "active", "time-sensitive", "critical"}
FCM_RESERVED_PREFIXES = ("google.", "gcm.")
FCM_RESERVED_KEYS = {"from", "notification", "message_type", "collapse_key"}

errors, warnings = [], []


def err(msg):
    errors.append(msg)


def warn(msg):
    warnings.append(msg)


def size_of(obj):
    return len(json.dumps(obj, separators=(",", ":"), ensure_ascii=False).encode("utf-8"))


def check_aps(payload, where, push_type, check_size=True):
    if not isinstance(payload, dict):
        err(f"{where}: payload must be a JSON object"); return
    aps = payload.get("aps")
    if not isinstance(aps, dict):
        err(f"{where}: missing 'aps' dictionary"); return
    if check_size:
        limit = APNS_VOIP_LIMIT if push_type == "voip" else APNS_LIMIT
        n = size_of(payload)
        if n > limit:
            err(f"{where}: payload is {n} bytes, APNs limit is {limit}")
        elif n > limit * 0.9:
            warn(f"{where}: payload is {n} bytes, close to the {limit}-byte limit")

    alert, content_available = aps.get("alert"), aps.get("content-available")
    if content_available not in (None, 1):
        err(f"{where}: aps.content-available must be 1 when present")
    if push_type == "background" or (content_available == 1 and not alert):
        for k in ("alert", "sound", "badge"):
            if k in aps:
                warn(f"{where}: background push should not include aps.{k}; send apns-push-type: background with apns-priority 5")
        if push_type != "background":
            warn(f"{where}: content-available only payload — send header apns-push-type: background and apns-priority: 5")
    if push_type == "alert" and not alert and not aps.get("badge") and not aps.get("sound"):
        err(f"{where}: alert push needs aps.alert, aps.badge or aps.sound")

    if alert is not None:
        if isinstance(alert, dict):
            known = {"title", "subtitle", "body", "launch-image", "title-loc-key", "title-loc-args",
                     "subtitle-loc-key", "subtitle-loc-args", "loc-key", "loc-args"}
            for k in alert:
                if k not in known:
                    warn(f"{where}: unknown aps.alert key '{k}'")
            if not any(k in alert for k in ("title", "body", "loc-key", "title-loc-key")):
                err(f"{where}: aps.alert has no title, body or loc-key")
            for k in ("loc-args", "title-loc-args", "subtitle-loc-args"):
                if k in alert and not (isinstance(alert[k], list) and all(isinstance(x, str) for x in alert[k])):
                    err(f"{where}: aps.alert.{k} must be an array of strings")
        elif not isinstance(alert, str):
            err(f"{where}: aps.alert must be a string or dictionary")

    if "badge" in aps and (not isinstance(aps["badge"], int) or aps["badge"] < 0):
        err(f"{where}: aps.badge must be a non-negative integer (0 clears the badge)")
    if "sound" in aps:
        s = aps["sound"]
        if isinstance(s, dict):
            if s.get("critical") == 1:
                warn(f"{where}: critical alerts need Apple's critical alerts entitlement")
            if "volume" in s and not (isinstance(s["volume"], (int, float)) and 0 <= s["volume"] <= 1):
                err(f"{where}: aps.sound.volume must be between 0 and 1")
        elif not isinstance(s, str):
            err(f"{where}: aps.sound must be a string or dictionary")
    if aps.get("mutable-content") not in (None, 1):
        err(f"{where}: aps.mutable-content must be 1 when present")
    if aps.get("mutable-content") == 1 and not alert:
        err(f"{where}: mutable-content requires a visible alert; the service extension never runs for silent pushes")
    if "interruption-level" in aps and aps["interruption-level"] not in INTERRUPTION_LEVELS:
        err(f"{where}: aps.interruption-level must be one of {sorted(INTERRUPTION_LEVELS)}")
    if aps.get("interruption-level") == "time-sensitive":
        warn(f"{where}: time-sensitive requires the Time Sensitive Notifications capability")
    if "relevance-score" in aps and not (isinstance(aps["relevance-score"], (int, float)) and 0 <= aps["relevance-score"] <= 1):
        err(f"{where}: aps.relevance-score must be between 0 and 1")
    for k in ("thread-id", "category", "target-content-id"):
        if k in aps and not isinstance(aps[k], str):
            err(f"{where}: aps.{k} must be a string")
    if push_type == "liveactivity":
        for k in ("event", "timestamp", "content-state"):
            if k not in aps:
                err(f"{where}: Live Activity push needs aps.{k}")
        if aps.get("event") not in (None, "start", "update", "end"):
            err(f"{where}: aps.event must be start, update or end")
    known_aps = {"alert", "badge", "sound", "content-available", "mutable-content", "category", "thread-id",
                 "target-content-id", "interruption-level", "relevance-score", "filter-criteria",
                 "stale-date", "content-state", "timestamp", "event", "dismissal-date", "attributes-type",
                 "attributes", "input-push-token", "input-push-channel"}
    for k in aps:
        if k not in known_aps:
            warn(f"{where}: unknown aps key '{k}' — custom data belongs outside 'aps'")
    for k, v in payload.items():
        if k != "aps" and isinstance(v, str) and any(t in k.lower() for t in ("password", "token", "secret", "ssn")):
            warn(f"{where}: custom key '{k}' looks sensitive; payloads are visible to the OS and in logs")


def check_fcm(body, push_type):
    msg = body.get("message")
    if not isinstance(msg, dict):
        err("FCM v1 body must be {\"message\": {...}} (legacy /fcm/send format was shut down)"); return
    targets = [k for k in ("token", "topic", "condition") if k in msg]
    if len(targets) != 1:
        err(f"message must have exactly one of token/topic/condition (found {targets or 'none'})")
    if "registration_ids" in msg or "to" in msg:
        err("'to'/'registration_ids' are legacy API fields; send one request per token or use topics")
    notif, data = msg.get("notification"), msg.get("data")
    if notif is not None:
        if not isinstance(notif, dict):
            err("message.notification must be an object")
        else:
            for k in notif:
                if k not in ("title", "body", "image"):
                    err(f"message.notification.{k} is not valid in v1; platform options go in android/apns/webpush")
    if data is not None:
        if not isinstance(data, dict):
            err("message.data must be an object")
        else:
            for k, v in data.items():
                if not isinstance(v, str):
                    err(f"message.data.{k} must be a string (stringify numbers, booleans and JSON)")
                if k in FCM_RESERVED_KEYS or k.startswith(FCM_RESERVED_PREFIXES):
                    err(f"message.data key '{k}' is reserved")
    n = size_of({k: msg[k] for k in ("notification", "data") if k in msg})
    if n > FCM_LIMIT:
        err(f"notification+data is {n} bytes, FCM limit is {FCM_LIMIT}")
    if notif is None and data is None and "apns" not in msg and "android" not in msg:
        err("message has no notification, data, android or apns content")

    android = msg.get("android")
    if isinstance(android, dict):
        if android.get("priority") not in (None, "high", "normal", "HIGH", "NORMAL"):
            err("android.priority must be high or normal")
        ttl = android.get("ttl")
        if ttl is not None and not (isinstance(ttl, str) and ttl.endswith("s")):
            err("android.ttl must be a duration string like \"3600s\"")
        an = android.get("notification")
        if isinstance(an, dict) and "channel_id" not in an:
            warn("android.notification.channel_id not set — Android 8+ falls back to the default channel")
        if notif is None and data is not None and android.get("priority") in ("high", "HIGH"):
            warn("high-priority data-only messages that never show a notification can be deprioritized by Android")

    apns = msg.get("apns")
    if isinstance(apns, dict):
        headers = apns.get("headers", {})
        if headers.get("apns-priority") not in (None, "5", "10"):
            err("apns.headers.apns-priority must be the string \"5\" or \"10\"")
        if "payload" in apns:
            check_aps(apns["payload"], "apns.payload", headers.get("apns-push-type", push_type), check_size=False)
            ca = apns["payload"].get("aps", {}).get("content-available") if isinstance(apns["payload"], dict) else None
            if ca == 1 and headers.get("apns-priority") == "10" and not apns["payload"].get("aps", {}).get("alert") and notif is None:
                err("background (content-available) push must use apns-priority \"5\"")
    elif notif is None and data is not None:
        warn("data-only message without apns block — iOS will not wake the app unless apns.payload.aps.content-available is 1")


def main():
    ap = argparse.ArgumentParser(description="Validate APNs / FCM v1 push payloads")
    ap.add_argument("file")
    ap.add_argument("--type", choices=["apns", "fcm"])
    ap.add_argument("--push-type", choices=["alert", "background", "voip", "liveactivity"], default=None)
    args = ap.parse_args()
    try:
        with open(args.file, encoding="utf-8") as fh:
            body = json.load(fh)
    except (OSError, json.JSONDecodeError) as e:
        print(f"error: {e}", file=sys.stderr); return 2
    kind = args.type or ("fcm" if isinstance(body, dict) and "message" in body else "apns")
    if kind == "fcm":
        check_fcm(body, args.push_type)
    else:
        check_aps(body, "apns", args.push_type)
    for w in warnings:
        print(f"WARN  {w}")
    for e in errors:
        print(f"ERROR {e}")
    print(f"{kind.upper()} payload: {len(errors)} error(s), {len(warnings)} warning(s)")
    return 1 if errors else 0


if __name__ == "__main__":
    sys.exit(main())