diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
index 0000000..4d05533
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,19 @@
+# Changelog
+
+All notable changes to this project are documented here. The format follows
+[Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
+
+## Unreleased
+
+### Added
+
+- `client.blog`: settings, keyword planning and articles under `/v1/blog`, with the
+ `BlogSettings`, `BlogKeyword`, `BlogArticle`, `BlogArticleFull` and `BlogArticleImage` models.
+- `fopost.webhooks.verify` (also exported as `verify_webhook`): verifies any FoPost webhook
+ delivery over the raw body and returns the `{event, data, timestamp}` envelope, with the
+ typed `ArticlePublishedPayload` for `article.published`.
+
+### Changed
+
+- `ChatAdapter.verify_webhook` now runs on the shared `fopost.webhooks` verifier. Behaviour and
+ error codes are unchanged.
diff --git a/README.md b/README.md
index 959d110..e9a2d9a 100644
--- a/README.md
+++ b/README.md
@@ -122,6 +122,44 @@ repurposed = client.ai.repurpose_url(
> `401` to an API key. They are here so the surface is complete once the server
> opens them up.
+## Blog
+
+`client.blog` plans keywords, writes articles and publishes them to a website. A post goes
+to a social network; an article goes to a website.
+
+```python
+article = client.blog.request_article(
+ workspace.id,
+ keyword="best trail running shoes",
+ instructions="Mention the return policy.",
+)
+print(article.status) # "queued"; it becomes a draft once written
+
+for draft in client.blog.list_articles(workspace.id, status="draft"):
+ client.blog.publish_article(workspace.id, draft.id)
+```
+
+With the Website target, every published article is POSTed to your `website_url` as a signed
+webhook. `verify_webhook` checks the signature over the raw body and returns the envelope:
+
+```python
+from fopost import ArticlePublishedPayload, verify_webhook
+
+
+@app.post("/webhooks/fopost")
+async def inbound(request: Request) -> Response:
+ secret = os.environ["FOPOST_WEBHOOK_SECRET"]
+ event = verify_webhook(await request.body(), request.headers, secret)
+ if event.event == "article.published":
+ article = ArticlePublishedPayload.model_validate(event.data)
+ save_article(article.slug, article.body_html)
+ return Response(status_code=204)
+```
+
+The secret is returned once, when `update_settings(..., website_url=...)` first saves the URL;
+`rotate_website_secret` issues a new one. A forged, stale or unsigned delivery raises
+`WebhookVerificationError` with a `code`.
+
## Analytics
```python
@@ -251,6 +289,7 @@ except FopostError as err:
| `knowledge` | `list`, `create`, `update`, `delete`, `sync`, `search` |
| `ads` | `list`, `external`, `boostable`, `connections`, `sources`, `providers`, `authorize`, `delete_connection`, `boost`, `create`, `refresh`, `set_status`, `delete`, `audiences`, `create_audience`, `search_targeting`, `lead_forms`, `create_lead_form`, `leads`, `account_tree`, `create_campaign`, `get_campaign`, `update_campaign`, `delete_campaign`, `duplicate_campaign`, `create_ad_set`, `get_ad_set`, `update_ad_set`, `delete_ad_set`, `duplicate_ad_set`, `create_network_ad`, `get_network_ad`, `update_network_ad`, `delete_network_ad`, `duplicate_network_ad`, `bulk_set_status`, `creatives`, `create_creative`, `get_creative`, `delete_creative`, `get_audience`, `update_audience`, `delete_audience`, `add_audience_users`, `add_audience_companies`, `estimate_reach`, `insights`, `ad_insights`, `get_lead_form`, `archive_lead_form`, `leads_feed`, `lead_pages`, `subscribe_lead_page`, `unsubscribe_lead_page`, `bid_pricing`, `supply_forecast`, `conversion_rules`, `create_conversion_rule`, `get_conversion_rule`, `update_conversion_rule`, `delete_conversion_rule`, `attach_conversion_rule`, `detach_conversion_rule`, `conversion_metrics`, `send_conversion_events` |
| `google_business` | `get_location`, `update_location`, `get_attributes`, `update_attributes`, `get_menus`, `replace_menus`, `get_services`, `replace_services`, `list_media`, `add_media`, `delete_media`, `list_place_actions`, `create_place_action`, `update_place_action`, `delete_place_action`, `get_verification_options`, `start_verification`, `complete_verification`, `get_performance`, `get_search_keywords`, `assign` |
+| `blog` | `get_settings`, `update_settings`, `rotate_website_secret`, `test_website`, `list_keywords`, `plan_keywords`, `add_keyword`, `update_keyword`, `delete_keyword`, `list_articles`, `request_article`, `get_article`, `update_article`, `regenerate_article`, `publish_article`, `reject_article`, `delete_article` |
| `validate` | `post`, `length`, `media` |
| `analytics` | `decay`, `frequency`, `timeline`, `changes`, `collect_post`, `native_posts` |
| `activity` | `list` |
diff --git a/src/fopost/__init__.py b/src/fopost/__init__.py
index f8b0f81..2ed1c24 100644
--- a/src/fopost/__init__.py
+++ b/src/fopost/__init__.py
@@ -18,6 +18,7 @@
from importlib.metadata import PackageNotFoundError
from importlib.metadata import version as _pkg_version
+from . import webhooks
from ._http import DEFAULT_BASE_URL
from .client import Fopost
from .errors import (
@@ -29,6 +30,9 @@
RateLimitError,
)
from .models import (
+ BLOG_ARTICLE_STATUSES,
+ BLOG_KEYWORD_STATUSES,
+ BLOG_TARGET_KINDS,
PLATFORMS,
POST_STATUSES,
AccountGroup,
@@ -57,10 +61,21 @@
AdSource,
AiCreditBalance,
AiCredits,
+ ArticlePublishedPayload,
Audience,
AudienceFilter,
AudiencesResult,
BidPricing,
+ BlogArticle,
+ BlogArticleFull,
+ BlogArticleImage,
+ BlogArticleStatus,
+ BlogKeyword,
+ BlogKeywordStatus,
+ BlogSettings,
+ BlogSettingsUpdate,
+ BlogTargetKind,
+ BlogWebsiteTest,
BlueskyLanguages,
BoostablePost,
Broadcast,
@@ -208,6 +223,8 @@
YouTubePlaylist,
YouTubeTranscript,
)
+from .webhooks import WebhookEvent, WebhookVerificationError
+from .webhooks import verify as verify_webhook
try:
__version__ = _pkg_version("fopost")
@@ -423,4 +440,22 @@
"ConversionRule",
"ReachEstimate",
"SupplyForecast",
+ "BLOG_ARTICLE_STATUSES",
+ "BLOG_KEYWORD_STATUSES",
+ "BLOG_TARGET_KINDS",
+ "ArticlePublishedPayload",
+ "BlogArticle",
+ "BlogArticleFull",
+ "BlogArticleImage",
+ "BlogArticleStatus",
+ "BlogKeyword",
+ "BlogKeywordStatus",
+ "BlogSettings",
+ "BlogSettingsUpdate",
+ "BlogTargetKind",
+ "BlogWebsiteTest",
+ "WebhookEvent",
+ "WebhookVerificationError",
+ "verify_webhook",
+ "webhooks",
]
diff --git a/src/fopost/chat_adapter.py b/src/fopost/chat_adapter.py
index b621153..eb7a63f 100644
--- a/src/fopost/chat_adapter.py
+++ b/src/fopost/chat_adapter.py
@@ -17,16 +17,19 @@
from __future__ import annotations
import builtins
-import hashlib
-import hmac
import json
-import time
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from typing import Any
from .client import Fopost
from .models import InboxAttachment, InboxItem
+from .webhooks import (
+ SIGNATURE_TOLERANCE_SECONDS,
+ WebhookVerificationError,
+ header,
+ verify_signature,
+)
__all__ = [
"INBOUND_EVENT",
@@ -42,12 +45,6 @@
#: The webhook event a new direct message raises.
INBOUND_EVENT = "inbox.message_received"
-#: Refuse a delivery signed longer ago than this. Matches the API's own tolerance.
-SIGNATURE_TOLERANCE_SECONDS = 300
-
-_SIGNATURE_HEADER = "x-fopost-signature"
-_TIMESTAMPED_SIGNATURE_HEADER = "x-fopost-signature-256"
-_TIMESTAMP_HEADER = "x-fopost-timestamp"
_EVENT_HEADER = "x-fopost-event"
@@ -126,26 +123,6 @@ def to_chat_message(item: InboxItem) -> ChatMessage:
)
-def _header(headers: Mapping[str, Any], name: str) -> str | None:
- getter = getattr(headers, "get", None)
- if getter is not None:
- value = getter(name)
- if value is None:
- value = getter(name.title())
- if value is not None:
- return value[0] if isinstance(value, (list, tuple)) else str(value)
- for key, value in dict(headers).items():
- if key.lower() == name:
- return value[0] if isinstance(value, (list, tuple)) else str(value)
- return None
-
-
-def _digest(value: str | None) -> str | None:
- if value is None:
- return None
- return value[7:] if value.startswith("sha256=") else value
-
-
class ChatAdapter:
"""Send and receive direct messages through one FoPost workspace."""
@@ -182,7 +159,7 @@ def parse_webhook(self, body: str | bytes, headers: Mapping[str, Any]) -> ChatEv
if not isinstance(envelope, dict):
raise ChatAdapterError("invalid_body", "Webhook body is not an object")
- event = envelope.get("event") or _header(headers, _EVENT_HEADER)
+ event = envelope.get("event") or header(headers, _EVENT_HEADER)
if event != INBOUND_EVENT:
raise ChatAdapterError(
"unexpected_event", f"Expected {INBOUND_EVENT}, got {event or 'nothing'}"
@@ -207,31 +184,10 @@ def verify_webhook(self, body: str | bytes, headers: Mapping[str, Any]) -> None:
"""Raise ``ChatAdapterError`` when the delivery is unsigned, forged or stale."""
if not self._webhook_secret:
raise ChatAdapterError("missing_secret", "Pass webhook_secret to verify a delivery")
-
- raw = body.encode() if isinstance(body, str) else body
- secret = self._webhook_secret.encode()
-
- timestamped = _digest(_header(headers, _TIMESTAMPED_SIGNATURE_HEADER))
- if timestamped is not None:
- try:
- sent_at = int(_header(headers, _TIMESTAMP_HEADER) or "")
- except ValueError as err:
- raise ChatAdapterError(
- "invalid_signature", "Signed delivery carries no timestamp"
- ) from err
- age = abs(time.time() - sent_at)
- if age > SIGNATURE_TOLERANCE_SECONDS:
- raise ChatAdapterError("stale_delivery", f"Delivery is {int(age)}s old")
- expected = hmac.new(secret, f"{sent_at}.".encode() + raw, hashlib.sha256).hexdigest()
- if not hmac.compare_digest(timestamped, expected):
- raise ChatAdapterError("invalid_signature", "Signature does not match the body")
- return
-
- plain = _digest(_header(headers, _SIGNATURE_HEADER))
- if plain is None:
- raise ChatAdapterError("invalid_signature", "Delivery carries no signature header")
- if not hmac.compare_digest(plain, hmac.new(secret, raw, hashlib.sha256).hexdigest()):
- raise ChatAdapterError("invalid_signature", "Signature does not match the body")
+ try:
+ verify_signature(body, headers, self._webhook_secret)
+ except WebhookVerificationError as err:
+ raise ChatAdapterError(err.code, err.message) from err
def receive(
self,
diff --git a/src/fopost/client.py b/src/fopost/client.py
index 6c97c3d..2e013b0 100644
--- a/src/fopost/client.py
+++ b/src/fopost/client.py
@@ -16,6 +16,7 @@
AdsResource,
AiResource,
AnalyticsResource,
+ BlogResource,
BroadcastsResource,
ContactsResource,
GoogleBusinessResource,
@@ -87,6 +88,7 @@ def __init__(
self.knowledge = KnowledgeResource(self._http)
self.activity = ActivityResource(self._http)
self.google_business = GoogleBusinessResource(self._http)
+ self.blog = BlogResource(self._http)
@property
def base_url(self) -> str:
diff --git a/src/fopost/models.py b/src/fopost/models.py
index 4f09263..81079ee 100644
--- a/src/fopost/models.py
+++ b/src/fopost/models.py
@@ -8,7 +8,7 @@
from __future__ import annotations
-from datetime import datetime
+from datetime import date, datetime
from typing import Any, Generic, Literal, TypeVar
from pydantic import AliasChoices, AliasGenerator, BaseModel, ConfigDict
@@ -162,6 +162,20 @@
"ReachFrequencyPrediction",
"ReachFrequencyResult",
"ValueRuleSet",
+ "BLOG_ARTICLE_STATUSES",
+ "BLOG_KEYWORD_STATUSES",
+ "BLOG_TARGET_KINDS",
+ "ArticlePublishedPayload",
+ "BlogArticle",
+ "BlogArticleFull",
+ "BlogArticleImage",
+ "BlogArticleStatus",
+ "BlogKeyword",
+ "BlogKeywordStatus",
+ "BlogSettings",
+ "BlogSettingsUpdate",
+ "BlogTargetKind",
+ "BlogWebsiteTest",
]
#: Every platform the API can publish to. Model fields stay plain `str`, so a
@@ -2280,3 +2294,161 @@ class GoogleConversionAction(FopostModel):
type: str
counting_type: str | None = None
value_minor: int | None = None
+
+
+# ── Blog ──────────────────────────────────────────────────────────
+
+BlogArticleStatus = Literal[
+ "queued",
+ "writing",
+ "draft",
+ "scheduled",
+ "publishing",
+ "published",
+ "failed",
+ "rejected",
+]
+
+BLOG_ARTICLE_STATUSES: tuple[str, ...] = (
+ "queued",
+ "writing",
+ "draft",
+ "scheduled",
+ "publishing",
+ "published",
+ "failed",
+ "rejected",
+)
+
+BlogKeywordStatus = Literal["suggested", "planned", "written", "dismissed"]
+
+BLOG_KEYWORD_STATUSES: tuple[str, ...] = ("suggested", "planned", "written", "dismissed")
+
+BlogTargetKind = Literal["product_blog", "account", "website"]
+
+BLOG_TARGET_KINDS: tuple[str, ...] = ("product_blog", "account", "website")
+
+
+class BlogSettings(FopostModel):
+ """How a workspace plans, writes and publishes articles."""
+
+ site_url: str | None = None
+ topic: str | None = None
+ audience: str | None = None
+ language: str = "en"
+ location_code: int | None = None
+ #: Articles planned per week; 0 turns planning off.
+ cadence_per_week: int = 0
+ #: Days a draft waits for review before it may auto-publish.
+ review_hold_days: int = 0
+ auto_publish: bool = False
+ image_tier: str | None = None
+ in_body_images: int | None = None
+ min_words: int | None = None
+ max_words: int | None = None
+ brand_voice_id: str | None = None
+ #: ``product_blog``, ``account`` or ``website``.
+ target_kind: str | None = None
+ target_account_id: str | None = None
+ target_platform: str | None = None
+ website_url: str | None = None
+ has_website_secret: bool = False
+ keyword_data_available: bool = False
+ updated_at: datetime | None = None
+
+
+class BlogSettingsUpdate(FopostModel):
+ """The saved settings, plus the website secret the one time it is issued."""
+
+ settings: BlogSettings
+ website_secret: str | None = None
+
+
+class BlogWebsiteTest(FopostModel):
+ """Outcome of a test delivery to the website target."""
+
+ ok: bool = False
+ status: int | None = None
+ error: str | None = None
+
+
+class BlogKeyword(FopostModel):
+ id: str
+ keyword: str
+ search_volume: int | None = None
+ #: 0 to 100.
+ difficulty: int | None = None
+ cpc: float | None = None
+ intent: str | None = None
+ #: ``provider``, ``user`` or ``agent``.
+ source: str = "user"
+ #: ``suggested``, ``planned``, ``written`` or ``dismissed``.
+ status: str = "suggested"
+ planned_for: date | None = None
+ created_at: datetime | None = None
+
+
+class BlogArticleImage(FopostModel):
+ media_id: str
+ url: str
+ alt: str = ""
+ #: Index of the outline heading the image sits under; absent on the featured image.
+ position: int | None = None
+
+
+class BlogArticle(FopostModel):
+ """An article as a listing returns it, without the body."""
+
+ id: str
+ title: str
+ slug: str
+ keyword: str | None = None
+ excerpt: str | None = None
+ meta_title: str | None = None
+ meta_description: str | None = None
+ featured_image: BlogArticleImage | None = None
+ tags: list[str] = []
+ word_count: int = 0
+ #: One of ``BLOG_ARTICLE_STATUSES``.
+ status: str = "queued"
+ #: ``plan``, ``user`` or ``agent``.
+ origin: str = "user"
+ review_until: datetime | None = None
+ publish_at: datetime | None = None
+ published_at: datetime | None = None
+ published_url: str | None = None
+ error: str | None = None
+ created_at: datetime | None = None
+ updated_at: datetime | None = None
+
+
+class BlogArticleFull(BlogArticle):
+ """One article with its body, outline, brief and images."""
+
+ instructions: str | None = None
+ outline: list[str] = []
+ #: The research the writer worked from.
+ brief: dict[str, Any] | None = None
+ body_markdown: str = ""
+ body_html: str = ""
+ images: list[BlogArticleImage] = []
+
+
+class ArticlePublishedPayload(FopostModel):
+ """``data`` of the ``article.published`` webhook. Image URLs expire after six hours."""
+
+ id: str
+ title: str
+ slug: str
+ excerpt: str | None = None
+ meta_title: str | None = None
+ meta_description: str | None = None
+ body_markdown: str = ""
+ body_html: str = ""
+ featured_image: BlogArticleImage | None = None
+ images: list[BlogArticleImage] = []
+ tags: list[str] = []
+ keyword: str | None = None
+ word_count: int = 0
+ published_at: datetime | None = None
+ published_url: str | None = None
diff --git a/src/fopost/resources/__init__.py b/src/fopost/resources/__init__.py
index 82fb15e..674a6f8 100644
--- a/src/fopost/resources/__init__.py
+++ b/src/fopost/resources/__init__.py
@@ -4,6 +4,7 @@
from .ads import AdsResource
from .ai import AiResource
from .analytics import AnalyticsResource
+from .blog import BlogResource
from .broadcasts import BroadcastsResource, SequencesResource
from .contacts import ContactsResource
from .google_business import GoogleBusinessResource
@@ -22,6 +23,7 @@
"ActivityResource",
"AdsResource",
"AiResource",
+ "BlogResource",
"BroadcastsResource",
"ContactsResource",
"SequencesResource",
diff --git a/src/fopost/resources/blog.py b/src/fopost/resources/blog.py
new file mode 100644
index 0000000..b8daa8b
--- /dev/null
+++ b/src/fopost/resources/blog.py
@@ -0,0 +1,261 @@
+"""``client.blog`` — plan keywords, request articles and publish them to a website.
+
+A post goes to a social network; an article goes to a website. Everything here
+is an article. Planning, requesting and regenerating spend AI credits and need
+the ``ai`` scope; publishing needs ``publish``; the rest needs ``posts``.
+
+These routes take ``workspaceId`` in camelCase, as a query parameter on reads
+and per-id routes and as a body field on the collection ``POST``/``PATCH``
+routes. Request fields are camelCase on the wire too; pass them snake_case here.
+"""
+
+from __future__ import annotations
+
+import builtins
+from collections.abc import Sequence
+from typing import Any
+
+from .._http import unwrap
+from ..models import (
+ BlogArticle,
+ BlogArticleFull,
+ BlogKeyword,
+ BlogSettings,
+ BlogSettingsUpdate,
+ BlogWebsiteTest,
+ _to_camel,
+)
+from ._base import UNSET, Resource, drop_unset, parse_list
+
+__all__ = ["BlogResource"]
+
+
+def _camel_body(fields: dict[str, Any]) -> dict[str, Any]:
+ return {_to_camel(key): value for key, value in drop_unset(fields).items()}
+
+
+def _statuses(status: str | Sequence[str] | None) -> str | None:
+ if status is None or isinstance(status, str):
+ return status
+ return ",".join(status)
+
+
+class BlogResource(Resource):
+ # ── Settings ──────────────────────────────────────────────────
+
+ def get_settings(self, workspace_id: str) -> BlogSettings:
+ body = self._http.get("/blog/settings", {"workspaceId": workspace_id})
+ return BlogSettings.model_validate(unwrap(body))
+
+ def update_settings(
+ self,
+ workspace_id: str,
+ *,
+ site_url: str | None | Any = UNSET,
+ topic: str | None | Any = UNSET,
+ audience: str | None | Any = UNSET,
+ language: str | Any = UNSET,
+ location_code: int | Any = UNSET,
+ cadence_per_week: int | Any = UNSET,
+ review_hold_days: int | Any = UNSET,
+ auto_publish: bool | Any = UNSET,
+ image_tier: str | Any = UNSET,
+ in_body_images: int | Any = UNSET,
+ min_words: int | Any = UNSET,
+ max_words: int | Any = UNSET,
+ brand_voice_id: str | None | Any = UNSET,
+ target_kind: str | None | Any = UNSET,
+ target_account_id: str | None | Any = UNSET,
+ website_url: str | None | Any = UNSET,
+ ) -> BlogSettingsUpdate:
+ """Patch the settings; an argument left out keeps its value.
+
+ ``website_secret`` on the result is set only the first time
+ ``website_url`` is saved. Store it then; it is never shown again.
+ """
+ body = _camel_body(
+ {
+ "site_url": site_url,
+ "topic": topic,
+ "audience": audience,
+ "language": language,
+ "location_code": location_code,
+ "cadence_per_week": cadence_per_week,
+ "review_hold_days": review_hold_days,
+ "auto_publish": auto_publish,
+ "image_tier": image_tier,
+ "in_body_images": in_body_images,
+ "min_words": min_words,
+ "max_words": max_words,
+ "brand_voice_id": brand_voice_id,
+ "target_kind": target_kind,
+ "target_account_id": target_account_id,
+ "website_url": website_url,
+ }
+ )
+ body["workspaceId"] = workspace_id
+ raw = self._http.patch("/blog/settings", body)
+ secret = raw.get("websiteSecret") if isinstance(raw, dict) else None
+ return BlogSettingsUpdate(
+ settings=BlogSettings.model_validate(unwrap(raw)), website_secret=secret
+ )
+
+ def rotate_website_secret(self, workspace_id: str) -> str:
+ """Issue a new website secret; the previous one stops verifying at once."""
+ body = self._http.post("/blog/settings/website-secret", {"workspaceId": workspace_id})
+ data = unwrap(body)
+ return str(data["websiteSecret"]) if isinstance(data, dict) else ""
+
+ def test_website(self, workspace_id: str) -> BlogWebsiteTest:
+ """Send a signed ``article.test`` delivery to the website target."""
+ body = self._http.post("/blog/settings/website-test", {"workspaceId": workspace_id})
+ return BlogWebsiteTest.model_validate(unwrap(body))
+
+ # ── Keywords ──────────────────────────────────────────────────
+
+ def list_keywords(
+ self,
+ workspace_id: str,
+ *,
+ status: str | Sequence[str] | None = None,
+ limit: int | None = None,
+ ) -> builtins.list[BlogKeyword]:
+ body = self._http.get(
+ "/blog/keywords",
+ {"workspaceId": workspace_id, "status": _statuses(status), "limit": limit},
+ )
+ return parse_list(BlogKeyword, unwrap(body))
+
+ def plan_keywords(
+ self, workspace_id: str, *, seeds: Sequence[str] | None = None
+ ) -> dict[str, Any]:
+ """Queue a research run; suggestions land in ``list_keywords`` as they arrive."""
+ payload: dict[str, Any] = {"workspaceId": workspace_id}
+ if seeds is not None:
+ payload["seeds"] = builtins.list(seeds)
+ data = unwrap(self._http.post("/blog/keywords/plan", payload))
+ return data if isinstance(data, dict) else {"status": data}
+
+ def add_keyword(
+ self, workspace_id: str, *, keyword: str, planned_for: str | None = None
+ ) -> BlogKeyword:
+ payload: dict[str, Any] = {"workspaceId": workspace_id, "keyword": keyword}
+ if planned_for is not None:
+ payload["plannedFor"] = planned_for
+ return BlogKeyword.model_validate(unwrap(self._http.post("/blog/keywords", payload)))
+
+ def update_keyword(
+ self,
+ workspace_id: str,
+ keyword_id: str,
+ *,
+ status: str | Any = UNSET,
+ planned_for: str | None | Any = UNSET,
+ ) -> BlogKeyword:
+ body = self._http.request(
+ "PATCH",
+ f"/blog/keywords/{keyword_id}",
+ json=_camel_body({"status": status, "planned_for": planned_for}),
+ params={"workspaceId": workspace_id},
+ )
+ return BlogKeyword.model_validate(unwrap(body))
+
+ def delete_keyword(self, workspace_id: str, keyword_id: str) -> None:
+ self._http.request(
+ "DELETE", f"/blog/keywords/{keyword_id}", params={"workspaceId": workspace_id}
+ )
+
+ # ── Articles ──────────────────────────────────────────────────
+
+ def list_articles(
+ self,
+ workspace_id: str,
+ *,
+ status: str | Sequence[str] | None = None,
+ limit: int | None = None,
+ ) -> builtins.list[BlogArticle]:
+ body = self._http.get(
+ "/blog/articles",
+ {"workspaceId": workspace_id, "status": _statuses(status), "limit": limit},
+ )
+ return parse_list(BlogArticle, unwrap(body))
+
+ def request_article(
+ self,
+ workspace_id: str,
+ *,
+ title: str | None = None,
+ keyword: str | None = None,
+ keyword_id: str | None = None,
+ instructions: str | None = None,
+ ) -> BlogArticle:
+ """Queue an article. One of ``title``, ``keyword`` or ``keyword_id`` is required."""
+ payload: dict[str, Any] = {"workspaceId": workspace_id}
+ for key, value in (
+ ("title", title),
+ ("keyword", keyword),
+ ("keywordId", keyword_id),
+ ("instructions", instructions),
+ ):
+ if value is not None:
+ payload[key] = value
+ return BlogArticle.model_validate(unwrap(self._http.post("/blog/articles", payload)))
+
+ def get_article(self, workspace_id: str, article_id: str) -> BlogArticleFull:
+ body = self._http.get(f"/blog/articles/{article_id}", {"workspaceId": workspace_id})
+ return BlogArticleFull.model_validate(unwrap(body))
+
+ def update_article(
+ self,
+ workspace_id: str,
+ article_id: str,
+ *,
+ title: str | Any = UNSET,
+ slug: str | Any = UNSET,
+ excerpt: str | None | Any = UNSET,
+ meta_title: str | None | Any = UNSET,
+ meta_description: str | None | Any = UNSET,
+ body_markdown: str | Any = UNSET,
+ tags: Sequence[str] | Any = UNSET,
+ publish_at: str | None | Any = UNSET,
+ ) -> BlogArticleFull:
+ """Edit a draft. ``publish_at`` puts it on the calendar; ``None`` returns it to review."""
+ body = self._http.request(
+ "PATCH",
+ f"/blog/articles/{article_id}",
+ json=_camel_body(
+ {
+ "title": title,
+ "slug": slug,
+ "excerpt": excerpt,
+ "meta_title": meta_title,
+ "meta_description": meta_description,
+ "body_markdown": body_markdown,
+ "tags": builtins.list(tags) if tags is not UNSET else UNSET,
+ "publish_at": publish_at,
+ }
+ ),
+ params={"workspaceId": workspace_id},
+ )
+ return BlogArticleFull.model_validate(unwrap(body))
+
+ def regenerate_article(self, workspace_id: str, article_id: str) -> BlogArticle:
+ return self._article_action(workspace_id, article_id, "regenerate")
+
+ def publish_article(self, workspace_id: str, article_id: str) -> BlogArticle:
+ """Publish now; a 409 ``no_target`` means the workspace has no target set."""
+ return self._article_action(workspace_id, article_id, "publish")
+
+ def reject_article(self, workspace_id: str, article_id: str) -> BlogArticle:
+ return self._article_action(workspace_id, article_id, "reject")
+
+ def delete_article(self, workspace_id: str, article_id: str) -> None:
+ self._http.request(
+ "DELETE", f"/blog/articles/{article_id}", params={"workspaceId": workspace_id}
+ )
+
+ def _article_action(self, workspace_id: str, article_id: str, action: str) -> BlogArticle:
+ body = self._http.request(
+ "POST", f"/blog/articles/{article_id}/{action}", params={"workspaceId": workspace_id}
+ )
+ return BlogArticle.model_validate(unwrap(body))
diff --git a/src/fopost/webhooks.py b/src/fopost/webhooks.py
new file mode 100644
index 0000000..3dd83ed
--- /dev/null
+++ b/src/fopost/webhooks.py
@@ -0,0 +1,157 @@
+"""Verify a FoPost webhook delivery and read its envelope.
+
+Every outbound FoPost webhook, including the Website blog target, is signed the
+same way: ``X-FoPost-Signature-256`` is an HMAC-SHA256 over ``{timestamp}.{body}``
+with ``X-FoPost-Timestamp`` beside it, and ``X-FoPost-Signature`` is the older
+HMAC over the body alone. Pass the raw request bytes, never a re-serialised
+object: a single reordered key changes the digest.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import time
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+__all__ = [
+ "SIGNATURE_TOLERANCE_SECONDS",
+ "WebhookEvent",
+ "WebhookVerificationError",
+ "verify",
+ "verify_signature",
+]
+
+#: Refuse a delivery signed longer ago than this. Matches the API's own tolerance.
+SIGNATURE_TOLERANCE_SECONDS = 300
+
+_SIGNATURE_HEADER = "x-fopost-signature"
+_TIMESTAMPED_SIGNATURE_HEADER = "x-fopost-signature-256"
+_TIMESTAMP_HEADER = "x-fopost-timestamp"
+_EVENT_HEADER = "x-fopost-event"
+
+
+class WebhookVerificationError(Exception):
+ """A delivery that could not be trusted.
+
+ ``code`` is one of ``missing_secret``, ``invalid_signature``, ``stale_delivery``
+ or ``invalid_body``.
+ """
+
+ def __init__(self, code: str, message: str) -> None:
+ super().__init__(message)
+ self.code = code
+ self.message = message
+
+
+@dataclass(frozen=True)
+class WebhookEvent:
+ """The envelope every FoPost webhook carries."""
+
+ event: str
+ data: dict[str, Any]
+ timestamp: str | None = None
+
+
+def header(headers: Mapping[str, Any], name: str) -> str | None:
+ """Read a header case-insensitively from a dict or a framework headers object."""
+ getter = getattr(headers, "get", None)
+ if getter is not None:
+ value = getter(name)
+ if value is None:
+ value = getter(name.title())
+ if value is not None:
+ return value[0] if isinstance(value, (list, tuple)) else str(value)
+ for key, value in dict(headers).items():
+ if key.lower() == name:
+ return value[0] if isinstance(value, (list, tuple)) else str(value)
+ return None
+
+
+def _digest(value: str | None) -> str | None:
+ if value is None:
+ return None
+ return value[7:] if value.startswith("sha256=") else value
+
+
+def verify_signature(
+ body: str | bytes,
+ headers: Mapping[str, Any],
+ secret: str,
+ *,
+ tolerance: float = SIGNATURE_TOLERANCE_SECONDS,
+) -> None:
+ """Raise ``WebhookVerificationError`` when the delivery is unsigned, forged or stale.
+
+ Prefers the timestamped ``X-FoPost-Signature-256`` and falls back to the
+ body-only ``X-FoPost-Signature`` when no timestamp header is present.
+ """
+ if not secret:
+ raise WebhookVerificationError("missing_secret", "A webhook secret is required")
+
+ raw = body.encode() if isinstance(body, str) else body
+ key = secret.encode()
+
+ timestamped = _digest(header(headers, _TIMESTAMPED_SIGNATURE_HEADER))
+ if timestamped is not None:
+ try:
+ sent_at = int(header(headers, _TIMESTAMP_HEADER) or "")
+ except ValueError as err:
+ raise WebhookVerificationError(
+ "invalid_signature", "Signed delivery carries no timestamp"
+ ) from err
+ age = abs(time.time() - sent_at)
+ if age > tolerance:
+ raise WebhookVerificationError("stale_delivery", f"Delivery is {int(age)}s old")
+ expected = hmac.new(key, f"{sent_at}.".encode() + raw, hashlib.sha256).hexdigest()
+ if not hmac.compare_digest(timestamped, expected):
+ raise WebhookVerificationError("invalid_signature", "Signature does not match the body")
+ return
+
+ plain = _digest(header(headers, _SIGNATURE_HEADER))
+ if plain is None:
+ raise WebhookVerificationError("invalid_signature", "Delivery carries no signature header")
+ if not hmac.compare_digest(plain, hmac.new(key, raw, hashlib.sha256).hexdigest()):
+ raise WebhookVerificationError("invalid_signature", "Signature does not match the body")
+
+
+def verify(
+ body: str | bytes,
+ headers: Mapping[str, Any],
+ secret: str,
+ *,
+ tolerance: float = SIGNATURE_TOLERANCE_SECONDS,
+) -> WebhookEvent:
+ """Verify the signature and return the parsed ``{event, data, timestamp}`` envelope.
+
+ ::
+
+ from fopost import verify_webhook
+
+ event = verify_webhook(request.body, request.headers, secret)
+ if event.event == "article.published":
+ article = ArticlePublishedPayload.model_validate(event.data)
+ """
+ verify_signature(body, headers, secret, tolerance=tolerance)
+
+ raw = body.decode() if isinstance(body, bytes) else body
+ try:
+ envelope = json.loads(raw)
+ except ValueError as err:
+ raise WebhookVerificationError("invalid_body", "Webhook body is not JSON") from err
+ if not isinstance(envelope, dict):
+ raise WebhookVerificationError("invalid_body", "Webhook body is not an object")
+
+ event = envelope.get("event") or header(headers, _EVENT_HEADER)
+ if not isinstance(event, str) or not event:
+ raise WebhookVerificationError("invalid_body", "Webhook body names no event")
+ data = envelope.get("data")
+ timestamp = envelope.get("timestamp")
+ return WebhookEvent(
+ event=event,
+ data=data if isinstance(data, dict) else {},
+ timestamp=timestamp if isinstance(timestamp, str) else None,
+ )
diff --git a/tests/test_blog.py b/tests/test_blog.py
new file mode 100644
index 0000000..2ae50bd
--- /dev/null
+++ b/tests/test_blog.py
@@ -0,0 +1,302 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import httpx
+import pytest
+import respx
+
+from fopost import Fopost, FopostError, PaymentRequiredError
+from tests.conftest import BASE_URL
+
+SETTINGS_FIXTURE: dict[str, Any] = {
+ "siteUrl": "https://yourbrand.com",
+ "topic": "Running shoes",
+ "audience": "Weekend runners",
+ "language": "en",
+ "locationCode": 2840,
+ "cadencePerWeek": 2,
+ "reviewHoldDays": 3,
+ "autoPublish": False,
+ "imageTier": "standard",
+ "inBodyImages": 2,
+ "minWords": 900,
+ "maxWords": 1600,
+ "brandVoiceId": None,
+ "targetKind": "website",
+ "targetAccountId": None,
+ "targetPlatform": None,
+ "websiteUrl": "https://yourbrand.com/api/fopost",
+ "hasWebsiteSecret": True,
+ "keywordDataAvailable": True,
+ "updatedAt": "2026-10-10T09:00:00.000Z",
+}
+
+KEYWORD_FIXTURE: dict[str, Any] = {
+ "id": "kw_1",
+ "keyword": "best trail running shoes",
+ "searchVolume": 12000,
+ "difficulty": 41,
+ "cpc": 1.2,
+ "intent": "commercial",
+ "source": "provider",
+ "status": "planned",
+ "plannedFor": "2026-10-14",
+ "createdAt": "2026-10-10T09:00:00.000Z",
+}
+
+ARTICLE_FIXTURE: dict[str, Any] = {
+ "id": "art_1",
+ "title": "The Best Trail Running Shoes of the Year",
+ "slug": "best-trail-running-shoes",
+ "keyword": "best trail running shoes",
+ "excerpt": "What to look for on rock, mud and root.",
+ "metaTitle": None,
+ "metaDescription": None,
+ "featuredImage": {"mediaId": "med_1", "url": "https://cdn.test/f.jpg", "alt": "Trail"},
+ "tags": ["shoes", "trail"],
+ "wordCount": 1240,
+ "status": "draft",
+ "origin": "plan",
+ "reviewUntil": "2026-10-13T09:00:00.000Z",
+ "publishAt": None,
+ "publishedAt": None,
+ "publishedUrl": None,
+ "error": None,
+ "createdAt": "2026-10-10T09:00:00.000Z",
+ "updatedAt": "2026-10-10T10:00:00.000Z",
+}
+
+ARTICLE_FULL_FIXTURE: dict[str, Any] = {
+ **ARTICLE_FIXTURE,
+ "instructions": "Mention the return policy.",
+ "outline": ["Why fit matters", "Our picks"],
+ "brief": {"questions": ["Are trail shoes waterproof?"], "competitors": []},
+ "bodyMarkdown": "# The Best Trail Running Shoes\n\nFit first.",
+ "bodyHtml": "The Best Trail Running Shoes
Fit first.
",
+ "images": [
+ {"mediaId": "med_2", "url": "https://cdn.test/1.jpg", "alt": "Mud", "position": 1},
+ ],
+}
+
+
+def _json(request: httpx.Request) -> dict[str, Any]:
+ return dict(json.loads(request.content))
+
+
+# ── Settings ──────────────────────────────────────────────────────
+
+
+@respx.mock
+def test_settings_round_trip(client: Fopost) -> None:
+ get_route = respx.get(f"{BASE_URL}/blog/settings").mock(
+ return_value=httpx.Response(200, json={"data": SETTINGS_FIXTURE})
+ )
+ patch_route = respx.patch(f"{BASE_URL}/blog/settings").mock(
+ return_value=httpx.Response(
+ 200, json={"data": SETTINGS_FIXTURE, "websiteSecret": "whsec_once"}
+ )
+ )
+ rotate_route = respx.post(f"{BASE_URL}/blog/settings/website-secret").mock(
+ return_value=httpx.Response(200, json={"data": {"websiteSecret": "whsec_new"}})
+ )
+ test_route = respx.post(f"{BASE_URL}/blog/settings/website-test").mock(
+ return_value=httpx.Response(
+ 502, json={"data": {"ok": False, "status": 500, "error": "upstream failed"}}
+ )
+ )
+
+ settings = client.blog.get_settings("ws_1")
+ assert settings.cadence_per_week == 2
+ assert settings.target_kind == "website"
+ assert settings.has_website_secret is True
+ assert dict(get_route.calls.last.request.url.params) == {"workspaceId": "ws_1"}
+ assert get_route.calls.last.request.headers["X-API-Key"] == "osk_test_key"
+
+ updated = client.blog.update_settings(
+ "ws_1", website_url="https://yourbrand.com/api/fopost", cadence_per_week=2, topic=None
+ )
+ assert updated.website_secret == "whsec_once"
+ assert updated.settings.site_url == "https://yourbrand.com"
+ assert _json(patch_route.calls.last.request) == {
+ "workspaceId": "ws_1",
+ "websiteUrl": "https://yourbrand.com/api/fopost",
+ "cadencePerWeek": 2,
+ "topic": None,
+ }
+
+ assert client.blog.rotate_website_secret("ws_1") == "whsec_new"
+ assert _json(rotate_route.calls.last.request) == {"workspaceId": "ws_1"}
+
+ with pytest.raises(FopostError) as err:
+ client.blog.test_website("ws_1")
+ assert err.value.status == 502
+ assert _json(test_route.calls.last.request) == {"workspaceId": "ws_1"}
+
+
+@respx.mock
+def test_website_test_reports_the_outcome(client: Fopost) -> None:
+ respx.post(f"{BASE_URL}/blog/settings/website-test").mock(
+ return_value=httpx.Response(200, json={"data": {"ok": True, "status": 200, "error": None}})
+ )
+ result = client.blog.test_website("ws_1")
+ assert result.ok is True and result.status == 200
+
+
+# ── Keywords ──────────────────────────────────────────────────────
+
+
+@respx.mock
+def test_keywords_round_trip(client: Fopost) -> None:
+ list_route = respx.get(f"{BASE_URL}/blog/keywords").mock(
+ return_value=httpx.Response(200, json={"data": [KEYWORD_FIXTURE]})
+ )
+ plan_route = respx.post(f"{BASE_URL}/blog/keywords/plan").mock(
+ return_value=httpx.Response(202, json={"data": {"status": "queued"}})
+ )
+ add_route = respx.post(f"{BASE_URL}/blog/keywords").mock(
+ return_value=httpx.Response(201, json={"data": KEYWORD_FIXTURE})
+ )
+ patch_route = respx.patch(f"{BASE_URL}/blog/keywords/kw_1").mock(
+ return_value=httpx.Response(200, json={"data": {**KEYWORD_FIXTURE, "status": "dismissed"}})
+ )
+ delete_route = respx.delete(f"{BASE_URL}/blog/keywords/kw_1").mock(
+ return_value=httpx.Response(204)
+ )
+
+ keywords = client.blog.list_keywords("ws_1", status=["suggested", "planned"], limit=50)
+ assert keywords[0].search_volume == 12000
+ assert str(keywords[0].planned_for) == "2026-10-14"
+ assert dict(list_route.calls.last.request.url.params) == {
+ "workspaceId": "ws_1",
+ "status": "suggested,planned",
+ "limit": "50",
+ }
+
+ assert client.blog.plan_keywords("ws_1", seeds=["trail shoes"]) == {"status": "queued"}
+ assert _json(plan_route.calls.last.request) == {"workspaceId": "ws_1", "seeds": ["trail shoes"]}
+
+ added = client.blog.add_keyword("ws_1", keyword="trail shoes", planned_for="2026-10-14")
+ assert added.id == "kw_1"
+ assert _json(add_route.calls.last.request) == {
+ "workspaceId": "ws_1",
+ "keyword": "trail shoes",
+ "plannedFor": "2026-10-14",
+ }
+
+ dismissed = client.blog.update_keyword("ws_1", "kw_1", status="dismissed", planned_for=None)
+ assert dismissed.status == "dismissed"
+ assert dict(patch_route.calls.last.request.url.params) == {"workspaceId": "ws_1"}
+ assert _json(patch_route.calls.last.request) == {"status": "dismissed", "plannedFor": None}
+
+ assert client.blog.delete_keyword("ws_1", "kw_1") is None
+ assert dict(delete_route.calls.last.request.url.params) == {"workspaceId": "ws_1"}
+
+
+@respx.mock
+def test_planning_on_an_empty_balance_is_a_payment_error(client: Fopost) -> None:
+ respx.post(f"{BASE_URL}/blog/keywords/plan").mock(
+ return_value=httpx.Response(
+ 402,
+ json={
+ "error": "insufficient_credits",
+ "message": "Your balance ran out.",
+ "upgrade_url": "/settings/billing",
+ },
+ )
+ )
+ with pytest.raises(PaymentRequiredError) as err:
+ client.blog.plan_keywords("ws_1")
+ assert err.value.code == "insufficient_credits"
+ assert err.value.upgrade_url == "/settings/billing"
+
+
+# ── Articles ──────────────────────────────────────────────────────
+
+
+@respx.mock
+def test_articles_round_trip(client: Fopost) -> None:
+ list_route = respx.get(f"{BASE_URL}/blog/articles").mock(
+ return_value=httpx.Response(200, json={"data": [ARTICLE_FIXTURE]})
+ )
+ create_route = respx.post(f"{BASE_URL}/blog/articles").mock(
+ return_value=httpx.Response(202, json={"data": {**ARTICLE_FIXTURE, "status": "queued"}})
+ )
+ get_route = respx.get(f"{BASE_URL}/blog/articles/art_1").mock(
+ return_value=httpx.Response(200, json={"data": ARTICLE_FULL_FIXTURE})
+ )
+ patch_route = respx.patch(f"{BASE_URL}/blog/articles/art_1").mock(
+ return_value=httpx.Response(200, json={"data": ARTICLE_FULL_FIXTURE})
+ )
+ regenerate_route = respx.post(f"{BASE_URL}/blog/articles/art_1/regenerate").mock(
+ return_value=httpx.Response(202, json={"data": {**ARTICLE_FIXTURE, "status": "writing"}})
+ )
+ publish_route = respx.post(f"{BASE_URL}/blog/articles/art_1/publish").mock(
+ return_value=httpx.Response(202, json={"data": {**ARTICLE_FIXTURE, "status": "publishing"}})
+ )
+ reject_route = respx.post(f"{BASE_URL}/blog/articles/art_1/reject").mock(
+ return_value=httpx.Response(200, json={"data": {**ARTICLE_FIXTURE, "status": "rejected"}})
+ )
+ delete_route = respx.delete(f"{BASE_URL}/blog/articles/art_1").mock(
+ return_value=httpx.Response(204)
+ )
+
+ articles = client.blog.list_articles("ws_1", status="draft", limit=10)
+ assert articles[0].word_count == 1240
+ assert articles[0].featured_image is not None
+ assert articles[0].featured_image.media_id == "med_1"
+ assert dict(list_route.calls.last.request.url.params) == {
+ "workspaceId": "ws_1",
+ "status": "draft",
+ "limit": "10",
+ }
+
+ queued = client.blog.request_article(
+ "ws_1", keyword_id="kw_1", instructions="Mention the return policy."
+ )
+ assert queued.status == "queued"
+ assert _json(create_route.calls.last.request) == {
+ "workspaceId": "ws_1",
+ "keywordId": "kw_1",
+ "instructions": "Mention the return policy.",
+ }
+
+ full = client.blog.get_article("ws_1", "art_1")
+ assert full.body_markdown.startswith("# The Best")
+ assert full.images[0].position == 1
+ assert full.brief is not None and full.brief["questions"] == ["Are trail shoes waterproof?"]
+ assert dict(get_route.calls.last.request.url.params) == {"workspaceId": "ws_1"}
+
+ client.blog.update_article(
+ "ws_1", "art_1", title="New title", tags=["shoes"], publish_at="2026-10-15T09:00:00Z"
+ )
+ assert dict(patch_route.calls.last.request.url.params) == {"workspaceId": "ws_1"}
+ assert _json(patch_route.calls.last.request) == {
+ "title": "New title",
+ "tags": ["shoes"],
+ "publishAt": "2026-10-15T09:00:00Z",
+ }
+
+ assert client.blog.regenerate_article("ws_1", "art_1").status == "writing"
+ assert dict(regenerate_route.calls.last.request.url.params) == {"workspaceId": "ws_1"}
+ assert client.blog.publish_article("ws_1", "art_1").status == "publishing"
+ assert dict(publish_route.calls.last.request.url.params) == {"workspaceId": "ws_1"}
+ assert client.blog.reject_article("ws_1", "art_1").status == "rejected"
+ assert dict(reject_route.calls.last.request.url.params) == {"workspaceId": "ws_1"}
+ assert client.blog.delete_article("ws_1", "art_1") is None
+ assert dict(delete_route.calls.last.request.url.params) == {"workspaceId": "ws_1"}
+
+
+@respx.mock
+def test_publishing_without_a_target_is_a_conflict(client: Fopost) -> None:
+ respx.post(f"{BASE_URL}/blog/articles/art_1/publish").mock(
+ return_value=httpx.Response(
+ 409, json={"error": "no_target", "message": "Choose where articles go first."}
+ )
+ )
+ with pytest.raises(FopostError) as err:
+ client.blog.publish_article("ws_1", "art_1")
+ assert type(err.value) is FopostError
+ assert err.value.status == 409
+ assert err.value.code == "no_target"
diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py
new file mode 100644
index 0000000..6b67fb1
--- /dev/null
+++ b/tests/test_webhooks.py
@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import time
+from typing import Any
+
+import pytest
+
+from fopost import ArticlePublishedPayload, WebhookVerificationError, verify_webhook, webhooks
+
+SECRET = "whsec_test"
+
+PAYLOAD: dict[str, Any] = {
+ "id": "art_1",
+ "title": "The Best Trail Running Shoes of the Year",
+ "slug": "best-trail-running-shoes",
+ "excerpt": None,
+ "metaTitle": None,
+ "metaDescription": None,
+ "bodyMarkdown": "# Shoes",
+ "bodyHtml": "Shoes
",
+ "featuredImage": {"mediaId": "med_1", "url": "https://cdn.test/f.jpg", "alt": "Trail"},
+ "images": [{"mediaId": "med_2", "url": "https://cdn.test/1.jpg", "alt": "Mud", "position": 0}],
+ "tags": ["shoes"],
+ "keyword": "trail shoes",
+ "wordCount": 1240,
+ "publishedAt": "2026-10-10T09:00:00.000Z",
+ "publishedUrl": "https://yourbrand.com/blog/best-trail-running-shoes",
+}
+
+
+def _sign(message: bytes) -> str:
+ return "sha256=" + hmac.new(SECRET.encode(), message, hashlib.sha256).hexdigest()
+
+
+def delivery(sent_at: int | None = None) -> tuple[bytes, dict[str, str]]:
+ body = json.dumps(
+ {"event": "article.published", "data": PAYLOAD, "timestamp": "2026-10-10T09:00:01.000Z"}
+ ).encode()
+ sent_at = int(time.time()) if sent_at is None else sent_at
+ return body, {
+ "X-FoPost-Event": "article.published",
+ "X-FoPost-Delivery": "dlv_1",
+ "X-FoPost-Timestamp": str(sent_at),
+ "X-FoPost-Signature": _sign(body),
+ "X-FoPost-Signature-256": _sign(f"{sent_at}.".encode() + body),
+ }
+
+
+def test_a_timestamped_signature_passes_and_the_envelope_is_parsed() -> None:
+ body, headers = delivery()
+ event = verify_webhook(body, headers, SECRET)
+ assert event.event == "article.published"
+ assert event.timestamp == "2026-10-10T09:00:01.000Z"
+ article = ArticlePublishedPayload.model_validate(event.data)
+ assert article.slug == "best-trail-running-shoes"
+ assert article.featured_image is not None and article.featured_image.media_id == "med_1"
+ assert article.images[0].position == 0
+ assert article.published_url == "https://yourbrand.com/blog/best-trail-running-shoes"
+
+
+def test_the_module_path_is_the_same_helper() -> None:
+ body, headers = delivery()
+ assert webhooks.verify(body, headers, SECRET) == verify_webhook(body, headers, SECRET)
+
+
+def test_a_tampered_body_fails() -> None:
+ body, headers = delivery()
+ tampered = body.replace(b"best-trail", b"worst-trail")
+ with pytest.raises(WebhookVerificationError) as err:
+ verify_webhook(tampered, headers, SECRET)
+ assert err.value.code == "invalid_signature"
+
+
+def test_a_stale_timestamp_fails() -> None:
+ body, headers = delivery(sent_at=int(time.time()) - 301)
+ with pytest.raises(WebhookVerificationError) as err:
+ verify_webhook(body, headers, SECRET)
+ assert err.value.code == "stale_delivery"
+
+
+def test_the_body_only_signature_passes_without_a_timestamp() -> None:
+ body, headers = delivery()
+ del headers["X-FoPost-Signature-256"]
+ del headers["X-FoPost-Timestamp"]
+ assert verify_webhook(body, headers, SECRET).event == "article.published"
+
+
+def test_the_body_only_signature_is_ignored_when_the_timestamped_one_is_present() -> None:
+ body, headers = delivery()
+ headers["X-FoPost-Signature-256"] = _sign(b"0." + body)
+ with pytest.raises(WebhookVerificationError) as err:
+ verify_webhook(body, headers, SECRET)
+ assert err.value.code == "invalid_signature"
+
+
+def test_header_lookup_is_case_insensitive() -> None:
+ body, headers = delivery()
+ lowered = {key.lower(): value for key, value in headers.items()}
+ assert verify_webhook(body.decode(), lowered, SECRET).event == "article.published"
+
+
+def test_a_missing_secret_is_refused() -> None:
+ body, headers = delivery()
+ with pytest.raises(WebhookVerificationError) as err:
+ verify_webhook(body, headers, "")
+ assert err.value.code == "missing_secret"