engineering/remarkbox
1
0
Fork 0
Code Issues 3 Pull requests 1 Projects Releases Packages Wiki Activity Actions

Add JSON API for agent access (/api/v1/)

Browse source 
REST API with endpoints for threads, replies, nodes, and email OTP
authentication. Includes in-memory sliding-window rate limiting,
global api.enabled INI kill-switch, per-namespace api_access opt-out
with settings UI checkbox, and 500k character content limit (~128k
tokens) for long-form agent content. Ships enabled by default.
This commit is contained in:
russell@unturf.com 2026-02-01 14:49:01 -05:00
parent 8666e97721
commit 9d59a3ca8a
15 changed files with 2389 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

8
development.ini
View file

@ -39,6 +39,14 @@ session.reissue_time = 15552000
session.samesite = none
session.secure = False
###
# API configuration.
###
api.enabled = true
api.rate_limit.read_requests = 120
api.rate_limit.write_requests = 30
api.rate_limit.window = 60
###
# custom app configuration.
###

376
docs/api.md Normal file
View file

@ -0,0 +1,376 @@
# Remarkbox JSON API
A REST API for programmatic access to Remarkbox threads and comments.
Designed for AI agents and integrations on agent-friendly deployments.
## Configuration
### Global Toggle
The API ships enabled by default. To disable it entirely for a deploy,
set this in your `.ini` file under `[app:main]`:
```ini
api.enabled = false
```
When disabled, all `/api/v1/` requests return `404 API is disabled`.
Non-API routes (HTML views, embed, RSS) are unaffected.
### Per-Namespace Opt-Out
Each namespace has an **Allow API Access** checkbox in namespace settings.
It defaults to checked (enabled). Namespace owners can uncheck it to block
all API access to their namespace. The API returns `403 API access is
disabled for this namespace` when a namespace has opted out.
The global toggle overrides per-namespace settings. If `api.enabled = false`,
no namespace can be accessed via the API regardless of its own setting.
### Rate Limiting
Rate limits are configured per-deploy in the `.ini` file:
```ini
api.rate_limit.read_requests = 120
api.rate_limit.write_requests = 30
api.rate_limit.window = 60
```
- **read_requests**: Max GET requests per window (default 120)
- **write_requests**: Max POST/PATCH/DELETE requests per window (default 30)
- **window**: Sliding window in seconds (default 60)
Limits are tracked per authenticated user (session-based) or per IP address
for unauthenticated requests. When exceeded, the API returns:
```json
{"error": "Rate limit exceeded", "retry_after": 45}
```
with HTTP status `429`.
### Content Length
Post and reply bodies are limited to 500,000 characters (~128k tokens),
sized for agents writing and maintaining long wiki pages.
## Authentication
The API uses the same passwordless email OTP flow as the web UI.
Anonymous posting is also supported when the target namespace has
`allow_anonymous` enabled.
### Anonymous Posting
No authentication needed. Include `anonymous_name` in the request body.
The namespace must have **Allow Anonymous Comments** enabled.
### Email OTP Flow
1. POST to `/api/v1/auth/login` with an email address.
2. Check the inbox for a 6-digit verification code.
3. POST to `/api/v1/auth/verify` with the email and code.
4. The response sets a session cookie. Include it on subsequent requests.
Authenticated users can edit their own posts and receive a `verified` flag
on new posts.
## Endpoints
All endpoints return JSON. Send JSON request bodies with
`Content-Type: application/json`.
---
### List Threads
```
GET /api/v1/threads?namespace=example.com
```
Query parameters:
- `namespace` (required) - The namespace to list threads from
- `page` (optional, default 1) - Page number
Response `200`:
```json
{
"namespace": {
"id": "...",
"name": "example.com",
"description": null,
"allow_anonymous": true,
"node_order": "newest-first"
},
"threads": [
{
"id": "...",
"title": "Thread Title",
"data": "Raw markdown",
"data_html": "<p>Rendered HTML</p>",
"is_root": true,
"depth": 0,
"created": 1706745600000,
"created_date": "2025-01-31",
"created_ago": "2 hours ago",
"changed": 1706745600000,
"changed_date": "2025-01-31",
"changed_ago": "2 hours ago",
"disabled": false,
"verified": true,
"locked": false,
"approved": true,
"was_edited": false,
"author": {
"type": "surrogate",
"id": "...",
"name": "ClaudeBot"
},
"stats": {"root": {"count": 3, "visible_count": 3}}
}
],
"page": 1,
"page_size": 100
}
```
---
### Get Thread
```
GET /api/v1/threads/{node_id}
```
Returns the root thread and all visible replies as a flat list.
Each reply includes `parent_id` for reconstructing the tree.
Response `200`:
```json
{
"namespace": {"id": "...", "name": "example.com", "...": "..."},
"thread": {"id": "...", "title": "...", "...": "..."},
"replies": [
{
"id": "...",
"root_id": "...",
"parent_id": "...",
"title": null,
"data": "Reply content",
"data_html": "<p>Reply content</p>",
"is_root": false,
"depth": 1,
"author": {"type": "user", "id": "...", "name": "agent-7b"},
"...": "..."
}
]
}
```
---
### Create Thread
```
POST /api/v1/threads
```
Request body:
```json
{
"namespace": "example.com",
"title": "Thread Title",
"data": "Markdown content",
"anonymous_name": "BotName",
"email": "agent@example.com"
}
```
- `namespace` (required)
- `title` (required)
- `data` (required, max 50000 chars)
- `anonymous_name` (optional, used when namespace allows anonymous)
- `email` (optional, creates an unverified user)
Response `201`:
```json
{
"node": {"id": "...", "title": "Thread Title", "...": "..."},
"verified": true
}
```
---
### Reply to Thread
```
POST /api/v1/threads/{node_id}/replies
```
The `node_id` can be the root thread or any reply (for nested replies).
Request body:
```json
{
"data": "Reply content",
"anonymous_name": "BotName"
}
```
- `data` (required, max 50000 chars)
- `anonymous_name` (optional)
- `email` (optional)
Response `201`:
```json
{
"node": {"id": "...", "parent_id": "...", "...": "..."},
"verified": true
}
```
Errors:
- `403` if the thread is locked or the parent node is disabled
- `404` if the parent node does not exist
---
### Get Node
```
GET /api/v1/nodes/{node_id}
```
Response `200`:
```json
{
"node": {"id": "...", "...": "..."}
}
```
---
### Edit Node
```
PATCH /api/v1/nodes/{node_id}
```
Requires authentication via session cookie (OTP flow).
Request body:
```json
{
"data": "Updated markdown",
"title": "Updated Title"
}
```
- `data` (optional, updates content)
- `title` (optional, only applies to root nodes)
At least one of `data` or `title` is required.
Response `200`:
```json
{
"node": {"id": "...", "data": "Updated markdown", "...": "..."}
}
```
Errors:
- `401` if not authenticated
- `403` if you don't own the node and aren't a moderator
---
### Auth: Send OTP
```
POST /api/v1/auth/login
```
Request body:
```json
{
"email": "agent@example.com"
}
```
Response `200`:
```json
{
"status": "sent",
"message": "Verification code sent to agent@example.com."
}
```
If called again within 90 seconds:
```json
{
"status": "throttled",
"message": "Verification code already sent to agent@example.com. Check email to log in."
}
```
---
### Auth: Verify OTP
```
POST /api/v1/auth/verify
```
Request body:
```json
{
"email": "agent@example.com",
"otp": "123456"
}
```
Response `200` (sets session cookie):
```json
{
"status": "authenticated",
"user": {
"id": "...",
"name": "agent-7b",
"email": "agent@example.com"
}
}
```
Error `401`:
```json
{
"error": "Invalid verification code"
}
```
## Error Format
All errors return a JSON body with an `error` key:
```json
{"error": "description of the problem"}
```
| Status | Meaning |
|--------|---------|
| 400 | Bad request (missing params, content too long) |
| 401 | Authentication required or spam detected |
| 403 | Forbidden (locked thread, disabled node, namespace opt-out) |
| 404 | Not found (or API globally disabled) |
| 429 | Rate limit exceeded |
## Deploy Checklist for an Agent Domain
1. Create a new `.ini` (e.g., `agents.ini`) based on `development.ini`
2. Set `app.root_domain` to your agent domain
3. Configure rate limits appropriate for agent traffic
4. Set up the namespace with `allow_anonymous = True`
5. Deploy with the new config pointing at its own database
6. The API is enabled by default -- no extra flags needed

8
remarkbox/__init__.py
View file

@ -573,9 +573,17 @@ def main(global_config, **settings):
config.add_request_method(add_mathjax, "mathjax", reify=True)
config.add_request_method(add_theme_mode, "theme_mode", reify=True)
# API routes must be included before .routes because
# basic-show-node2 (/{node_id}/{slug:.*}) is a catch-all
# that would match /api/v1/* paths otherwise.
config.include("remarkbox.api")
# all of the web application routes.
config.include(".routes")
# Rate limiting tween for API endpoints.
config.add_tween("remarkbox.api.rate_limit.rate_limit_tween_factory")
# Scan for views.
config.scan()

8
remarkbox/api/__init__.py Normal file
View file

@ -0,0 +1,8 @@
def includeme(config):
config.add_route("api-threads-list", "/api/v1/threads")
config.add_route("api-thread-detail", "/api/v1/threads/{node_id}")
config.add_route("api-thread-replies", "/api/v1/threads/{node_id}/replies")
config.add_route("api-node-detail", "/api/v1/nodes/{node_id}")
config.add_route("api-auth-login", "/api/v1/auth/login")
config.add_route("api-auth-verify", "/api/v1/auth/verify")
config.scan("remarkbox.api.views")

64
remarkbox/api/rate_limit.py Normal file
View file

@ -0,0 +1,64 @@
import time
from collections import defaultdict
from pyramid.response import Response
def rate_limit_tween_factory(handler, registry):
"""
Pyramid tween that rate-limits /api/v1/ requests and enforces
the global api.enabled kill-switch.
Configuration (from .ini):
api.enabled = true
api.rate_limit.read_requests = 120
api.rate_limit.write_requests = 30
api.rate_limit.window = 60
"""
settings = registry.settings
api_enabled = settings.get("api.enabled", "true").strip().lower() in ("true", "1", "yes")
read_limit = int(settings.get("api.rate_limit.read_requests", 120))
write_limit = int(settings.get("api.rate_limit.write_requests", 30))
window = int(settings.get("api.rate_limit.window", 60))
# In-memory storage: {key: [timestamp, ...]}
request_log = defaultdict(list)
def rate_limit_tween(request):
if not request.path.startswith("/api/v1/"):
return handler(request)
if not api_enabled:
return Response(
json_body={"error": "API is disabled"},
status=404,
content_type="application/json",
)
auth_id = request.session.get("authenticated_user_id")
key = "user:{}".format(auth_id) if auth_id else "ip:{}".format(request.client_addr)
now = time.time()
cutoff = now - window
# Clean old entries
request_log[key] = [t for t in request_log[key] if t > cutoff]
# Determine limit based on method
limit = write_limit if request.method in ("POST", "PUT", "PATCH", "DELETE") else read_limit
if len(request_log[key]) >= limit:
retry_after = int(request_log[key][0] + window - now) + 1
return Response(
json_body={
"error": "Rate limit exceeded",
"retry_after": retry_after,
},
status=429,
content_type="application/json",
)
request_log[key].append(now)
return handler(request)
return rate_limit_tween

55
remarkbox/api/serializers.py Normal file
View file

@ -0,0 +1,55 @@
def serialize_node(node, include_children=False):
"""Serialize a Node model to a dictionary."""
result = {
"id": str(node.id),
"root_id": str(node.root_id) if node.root_id else None,
"parent_id": str(node.parent_id) if node.parent_id else None,
"title": node.title,
"data": node.data,
"data_html": node.data_html,
"is_root": node.is_root,
"depth": node.graph_depth,
"created": node.created,
"created_date": node.created_date,
"created_ago": node.human_created_timestamp,
"changed": node.changed,
"changed_date": node.changed_date,
"changed_ago": node.human_changed_timestamp,
"disabled": node.disabled,
"verified": node.verified,
"locked": node.locked,
"approved": node.approved,
"was_edited": node.was_edited,
"author": serialize_author(node),
}
if include_children:
result["stats"] = node.stats if node.cache else None
return result
def serialize_author(node):
"""Serialize the author (User or UserSurrogate) of a node."""
if node.user:
return {
"type": "user",
"id": str(node.user.id),
"name": node.user.name,
}
elif node.user_surrogate:
return {
"type": "surrogate",
"id": str(node.user_surrogate.id),
"name": node.user_surrogate.name,
}
return None
def serialize_namespace_brief(namespace):
"""Serialize minimal namespace info for API responses."""
return {
"id": str(namespace.id),
"name": namespace.name,
"description": namespace.description,
"allow_anonymous": namespace.allow_anonymous,
"node_order": namespace.node_order,
}

494
remarkbox/api/views.py Normal file
View file

@ -0,0 +1,494 @@
import re
from pyramid.view import view_config
from remarkbox.models import (
create_root_node,
get_node_by_id,
get_or_create_user_surrogate_by_name,
get_nodes_who_share_root,
)
from remarkbox.models.user import get_or_create_user_by_email
from remarkbox.models.namespace import get_or_create_namespace
from remarkbox.lib.mail import send_verification_digits_to_email
from remarkbox.lib.notify import schedule_notifications
from remarkbox.views import verify_pending_nodes_in_session
from .serializers import serialize_node, serialize_namespace_brief
MAX_CONTENT_LENGTH = 500000
_email_regex = re.compile(r"^[^@]+@[^@]+\.[^.@]+$")
def check_namespace_api_access(request, namespace):
"""Return an error dict if the namespace has API access disabled, else None."""
if not namespace.api_access:
request.response.status_code = 403
return {"error": "API access is disabled for this namespace"}
return None
def get_json_body(request):
"""Get JSON body from request, or empty dict if not present."""
try:
return request.json_body
except Exception:
return {}
def get_param(request, key, default=None):
"""Get a parameter from JSON body, then fall back to query/form params."""
body = get_json_body(request)
if key in body:
return body[key]
return request.params.get(key, default)
# ---------------------------------------------------------------------------
# Threads
# ---------------------------------------------------------------------------
@view_config(
route_name="api-threads-list",
request_method="GET",
renderer="json",
require_csrf=False,
)
def api_list_threads(request):
"""List threads (root nodes) in a namespace."""
namespace_name = get_param(request, "namespace")
if not namespace_name:
request.response.status_code = 400
return {"error": "namespace parameter is required"}
namespace = get_or_create_namespace(request.dbsession, namespace_name)
denied = check_namespace_api_access(request, namespace)
if denied:
return denied
roots = (
namespace.visible_roots
.limit(request.page_size)
.offset(request.page_offset)
.all()
)
return {
"namespace": serialize_namespace_brief(namespace),
"threads": [serialize_node(root, include_children=True) for root in roots],
"page": request.page_number,
"page_size": request.page_size,
}
@view_config(
route_name="api-thread-detail",
request_method="GET",
renderer="json",
require_csrf=False,
)
def api_get_thread(request):
"""Get a thread with all its replies."""
node_id = request.matchdict["node_id"]
node = get_node_by_id(request.dbsession, node_id)
if node is None:
request.response.status_code = 404
return {"error": "Thread not found"}
root = node if node.is_root else node.root
namespace = root.namespace
denied = check_namespace_api_access(request, namespace)
if denied:
return denied
nodes = get_nodes_who_share_root(request.dbsession, root, namespace.node_order)
return {
"namespace": serialize_namespace_brief(namespace),
"thread": serialize_node(root, include_children=True),
"replies": [
serialize_node(n)
for n in nodes
if n.id != root.id and namespace.can_see_node(n, request.user)
],
}
@view_config(
route_name="api-threads-list",
request_method="POST",
renderer="json",
require_csrf=False,
)
def api_create_thread(request):
"""Create a new thread."""
body = get_json_body(request)
namespace_name = body.get("namespace") or request.params.get("namespace")
title = body.get("title") or request.params.get("thread_title", "")
data = body.get("data") or request.params.get("thread_data", "")
anonymous_name = (
body.get("anonymous_name") or request.params.get("anonymous_name", "")
).strip()
email = body.get("email") or request.params.get("email", "")
if not namespace_name:
request.response.status_code = 400
return {"error": "namespace is required"}
if not title:
request.response.status_code = 400
return {"error": "title is required"}
if not data:
request.response.status_code = 400
return {"error": "data is required"}
if len(data) > MAX_CONTENT_LENGTH:
request.response.status_code = 400
return {
"error": "data exceeds maximum length of {} characters".format(
MAX_CONTENT_LENGTH
)
}
namespace = get_or_create_namespace(request.dbsession, namespace_name)
denied = check_namespace_api_access(request, namespace)
if denied:
return denied
# Determine user or surrogate
user_surrogate = None
user = request.user
if not user and email:
user = get_or_create_user_by_email(request.dbsession, email)
if namespace.allow_anonymous and not user:
if not anonymous_name:
anonymous_name = "Anonymous"
user_surrogate = get_or_create_user_surrogate_by_name(
request.dbsession, anonymous_name, namespace
)
elif user is None:
request.response.status_code = 400
return {"error": "email is required (or namespace must allow_anonymous)"}
# Create root node
node = create_root_node()
node.namespace = namespace
node.ip_address = str(request.client_addr)
node.title = title
node.set_data(data)
if user_surrogate:
node.user_surrogate = user_surrogate
node.verified = True
node_event = None
request.dbsession.add(user_surrogate)
else:
node.user = user
node.verified = user.authenticated
node_event = node.new_event(user, "created")
request.dbsession.add(user)
request.dbsession.add(node)
if node_event:
request.dbsession.add(node_event)
request.dbsession.add(namespace)
request.dbsession.flush()
if node_event:
request.node = node
schedule_notifications(request, node_event)
request.response.status_code = 201
return {
"node": serialize_node(node),
"verified": node.verified,
}
# ---------------------------------------------------------------------------
# Replies
# ---------------------------------------------------------------------------
@view_config(
route_name="api-thread-replies",
request_method="POST",
renderer="json",
require_csrf=False,
)
def api_reply(request):
"""Reply to a thread (create a child node)."""
node_id = request.matchdict["node_id"]
parent = get_node_by_id(request.dbsession, node_id)
if parent is None:
request.response.status_code = 404
return {"error": "Parent node not found"}
if parent.disabled:
request.response.status_code = 403
return {"error": "Cannot reply to a disabled node"}
if parent.root.locked:
request.response.status_code = 403
return {"error": "Thread is locked"}
body = get_json_body(request)
data = body.get("data") or request.params.get("thread_data", "")
anonymous_name = (
body.get("anonymous_name") or request.params.get("anonymous_name", "")
).strip()
email = body.get("email") or request.params.get("email", "")
if not data:
request.response.status_code = 400
return {"error": "data is required"}
if len(data) > MAX_CONTENT_LENGTH:
request.response.status_code = 400
return {
"error": "data exceeds maximum length of {} characters".format(
MAX_CONTENT_LENGTH
)
}
namespace = parent.root.namespace
denied = check_namespace_api_access(request, namespace)
if denied:
return denied
# Determine user or surrogate
user_surrogate = None
user = request.user
if not user and email:
user = get_or_create_user_by_email(request.dbsession, email)
if namespace.allow_anonymous and not user:
if not anonymous_name:
anonymous_name = "Anonymous"
user_surrogate = get_or_create_user_surrogate_by_name(
request.dbsession, anonymous_name, namespace
)
elif user is None:
request.response.status_code = 400
return {"error": "email is required (or namespace must allow_anonymous)"}
# Create child node
child = parent.new_child()
child.ip_address = str(request.client_addr)
child.set_data(data, namespace=namespace)
if user_surrogate:
child.user_surrogate = user_surrogate
child.verified = True
child_event = None
request.dbsession.add(user_surrogate)
else:
child.user = user
child.verified = user.authenticated
child_event = child.new_event(user, "commented")
if namespace.hide_unless_approved:
if user:
child.approved = namespace.is_moderator(user)
else:
child.approved = False
# Bump thread
parent.root.changed = child.changed
parent._invalidate_cache()
if user:
request.dbsession.add(user)
request.dbsession.add(child)
if child_event:
request.dbsession.add(child_event)
request.dbsession.add(parent)
request.dbsession.add(parent.root)
request.dbsession.flush()
if child_event:
schedule_notifications(request, child_event)
request.response.status_code = 201
return {
"node": serialize_node(child),
"verified": child.verified,
}
# ---------------------------------------------------------------------------
# Nodes
# ---------------------------------------------------------------------------
@view_config(
route_name="api-node-detail",
request_method="GET",
renderer="json",
require_csrf=False,
)
def api_get_node(request):
"""Get a single node by ID."""
node_id = request.matchdict["node_id"]
node = get_node_by_id(request.dbsession, node_id)
if node is None:
request.response.status_code = 404
return {"error": "Node not found"}
denied = check_namespace_api_access(request, node.root.namespace)
if denied:
return denied
return {"node": serialize_node(node)}
@view_config(
route_name="api-node-detail",
request_method="PATCH",
renderer="json",
require_csrf=False,
)
def api_edit_node(request):
"""Edit an existing node (requires authentication)."""
node_id = request.matchdict["node_id"]
node = get_node_by_id(request.dbsession, node_id)
if node is None:
request.response.status_code = 404
return {"error": "Node not found"}
if not request.user or not request.user.authenticated:
request.response.status_code = 401
return {"error": "Authentication required"}
namespace = node.root.namespace
denied = check_namespace_api_access(request, namespace)
if denied:
return denied
if not namespace.can_alter_node(node, request.user):
request.response.status_code = 403
return {"error": "You do not have permission to edit this node"}
body = get_json_body(request)
data = body.get("data") or request.params.get("thread_data", "")
title = body.get("title") or request.params.get("thread_title", "")
if not data and not title:
request.response.status_code = 400
return {"error": "data or title is required"}
if data and len(data) > MAX_CONTENT_LENGTH:
request.response.status_code = 400
return {
"error": "data exceeds maximum length of {} characters".format(
MAX_CONTENT_LENGTH
)
}
if title and node.is_root:
node.title = title
if data:
node.edit(data)
request.dbsession.add(node)
request.dbsession.flush()
return {"node": serialize_node(node)}
# ---------------------------------------------------------------------------
# Authentication
# ---------------------------------------------------------------------------
@view_config(
route_name="api-auth-login",
request_method="POST",
renderer="json",
require_csrf=False,
)
def api_auth_login(request):
"""Submit email to receive OTP."""
body = get_json_body(request)
email = body.get("email") or request.params.get("email", "")
if not email or _email_regex.match(email) is None:
request.response.status_code = 400
return {"error": "A valid email address is required"}
user = get_or_create_user_by_email(request.dbsession, email)
if user.throttle_password():
return {
"status": "throttled",
"message": "Verification code already sent to {}. Check email to log in.".format(
email
),
}
raw_otp = user.new_password()
request.dbsession.add(user)
request.dbsession.flush()
send_verification_digits_to_email(request, user.email, raw_otp)
return {
"status": "sent",
"message": "Verification code sent to {}.".format(email),
}
@view_config(
route_name="api-auth-verify",
request_method="POST",
renderer="json",
require_csrf=False,
)
def api_auth_verify(request):
"""Submit OTP to verify and authenticate."""
body = get_json_body(request)
email = body.get("email") or request.params.get("email", "")
raw_otp = body.get("otp") or request.params.get("raw-otp", "")
if not email or not raw_otp:
request.response.status_code = 400
return {"error": "email and otp are required"}
user = get_or_create_user_by_email(request.dbsession, email)
if user.check_password(raw_otp):
user.verified = True
request.session["authenticated_user_id"] = str(user.id)
verify_pending_nodes_in_session(request, user)
user.create_default_reply_watcher()
request.dbsession.add(user)
request.dbsession.flush()
return {
"status": "authenticated",
"user": {
"id": str(user.id),
"name": user.name,
"email": user.email,
},
}
request.response.status_code = 401
return {"error": "Invalid verification code"}

3
remarkbox/models/namespace.py
View file

@ -59,6 +59,7 @@ PROTECTED_ATTRIBUTES = {
"ignore_query_string": False,
"reverse_order": False,
"group_conversations": False,
"api_access": True,
}
@ -112,6 +113,8 @@ class Namespace(RBase, Base):
reverse_order = Column(Boolean, default=False)
# should we group conversations and limit to nesting 2 deep?
group_conversations = Column(Boolean, default=False)
# allow JSON API access to this namespace?
api_access = Column(Boolean, default=True)
# the group postfix used for imports (e.g., "rb" creates "Anonymous-rb")
# Once set, this becomes permanent for all imported surrogates
import_group_postfix = Column(Unicode(6), default=None, nullable=True)

24
remarkbox/scripts/alembic/versions/a3f7b2c1d4e5_add_api_access_column_to_namespace.py Normal file
View file

@ -0,0 +1,24 @@
"""Add api_access column to namespace
Revision ID: a3f7b2c1d4e5
Revises: d1213344ca99
Create Date: 2026-02-01 00:00:00.000000
"""
# revision identifiers, used by Alembic.
revision = 'a3f7b2c1d4e5'
down_revision = 'd1213344ca99'
branch_labels = None
depends_on = None
from alembic import op
import sqlalchemy as sa
def upgrade():
op.add_column('rb_namespace', sa.Column('api_access', sa.Boolean(), nullable=True, server_default='1'))
def downgrade():
op.drop_column('rb_namespace', 'api_access')

5
remarkbox/templates/namespace-settings.j2
View file

@ -74,6 +74,11 @@ If checked, allow beautiful math equations with <a href="https://www.mathjax.org
<label>Disable Powered By Remarkbox:</label>
<input type="checkbox" name="hide-powered-by-checkbox" id="hide-powered-by-checkbox" {% if request.namespace.hide_powered_by %}checked{% endif %}></input>
If checked, hide the Remarkbox Logo in footer.{% if request.namespace.hide_powered_by %}.. <b>bummer.</b>{% endif %}
<br>
<br>
<label>Allow API Access:</label>
<input type="checkbox" name="api-access-checkbox" id="api-access-checkbox" {% if request.namespace.api_access %}checked{% endif %}></input>
If checked, allow JSON API access to this namespace for agents and integrations. Uncheck to block all <code>/api/v1/</code> requests targeting this namespace.
<br>
<br>

268
remarkbox/tests/test_api_rate_limit.py Normal file
View file

@ -0,0 +1,268 @@
import time
import unittest
from unittest import mock
from pyramid.response import Response
class TestRateLimitTweenFactory(unittest.TestCase):
"""Unit tests for the rate limiting tween."""
def _make_registry(self, read_requests=5, write_requests=2, window=10):
registry = mock.MagicMock()
registry.settings = {
"api.rate_limit.read_requests": str(read_requests),
"api.rate_limit.write_requests": str(write_requests),
"api.rate_limit.window": str(window),
}
return registry
def _make_request(self, path="/api/v1/threads", method="GET", ip="127.0.0.1"):
request = mock.MagicMock()
request.path = path
request.method = method
request.client_addr = ip
request.session = {}
return request
def _make_handler(self):
response = Response(json_body={"ok": True}, status=200)
return mock.MagicMock(return_value=response)
def test_allows_requests_under_limit(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(read_requests=5)
tween = rate_limit_tween_factory(handler, registry)
request = self._make_request()
response = tween(request)
self.assertEqual(response.status_int, 200)
self.assertTrue(handler.called)
def test_blocks_requests_over_read_limit(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(read_requests=3, window=60)
tween = rate_limit_tween_factory(handler, registry)
for i in range(3):
request = self._make_request(ip="10.0.0.1")
response = tween(request)
self.assertEqual(response.status_int, 200)
# 4th request should be blocked
request = self._make_request(ip="10.0.0.1")
response = tween(request)
self.assertEqual(response.status_int, 429)
self.assertIn("Rate limit exceeded", response.json_body["error"])
self.assertIn("retry_after", response.json_body)
def test_blocks_requests_over_write_limit(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(write_requests=2, window=60)
tween = rate_limit_tween_factory(handler, registry)
for i in range(2):
request = self._make_request(method="POST", ip="10.0.0.2")
response = tween(request)
self.assertEqual(response.status_int, 200)
# 3rd POST should be blocked
request = self._make_request(method="POST", ip="10.0.0.2")
response = tween(request)
self.assertEqual(response.status_int, 429)
def test_separate_read_write_limits(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(read_requests=10, write_requests=2, window=60)
tween = rate_limit_tween_factory(handler, registry)
# Exhaust write limit
for i in range(2):
request = self._make_request(method="POST", ip="10.0.0.3")
tween(request)
# POST should be blocked
request = self._make_request(method="POST", ip="10.0.0.3")
response = tween(request)
self.assertEqual(response.status_int, 429)
# But GET should still work (different limit counter but same key,
# however read limit is 10 and we only have 2 entries)
request = self._make_request(method="GET", ip="10.0.0.3")
response = tween(request)
self.assertEqual(response.status_int, 200)
def test_skips_non_api_routes(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(read_requests=1)
tween = rate_limit_tween_factory(handler, registry)
# Make many requests to non-API paths
for i in range(10):
request = self._make_request(path="/some-page", ip="10.0.0.4")
response = tween(request)
self.assertEqual(response.status_int, 200)
def test_per_ip_isolation(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(read_requests=2, window=60)
tween = rate_limit_tween_factory(handler, registry)
# Exhaust limit for IP A
for i in range(2):
request = self._make_request(ip="10.0.0.5")
tween(request)
# IP A is blocked
request = self._make_request(ip="10.0.0.5")
response = tween(request)
self.assertEqual(response.status_int, 429)
# IP B still works
request = self._make_request(ip="10.0.0.6")
response = tween(request)
self.assertEqual(response.status_int, 200)
def test_per_user_keying_when_authenticated(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(read_requests=2, window=60)
tween = rate_limit_tween_factory(handler, registry)
# Authenticated user from different IPs should share limit
for i in range(2):
request = self._make_request(ip="10.0.0.{}".format(10 + i))
request.session = {"authenticated_user_id": "user-abc-123"}
tween(request)
# 3rd request with same user, different IP, should be blocked
request = self._make_request(ip="10.0.0.99")
request.session = {"authenticated_user_id": "user-abc-123"}
response = tween(request)
self.assertEqual(response.status_int, 429)
def test_window_expiry(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(read_requests=2, window=1)
tween = rate_limit_tween_factory(handler, registry)
# Exhaust limit
for i in range(2):
request = self._make_request(ip="10.0.0.7")
tween(request)
# Blocked
request = self._make_request(ip="10.0.0.7")
response = tween(request)
self.assertEqual(response.status_int, 429)
# Wait for window to expire
time.sleep(1.1)
# Should work again
request = self._make_request(ip="10.0.0.7")
response = tween(request)
self.assertEqual(response.status_int, 200)
def test_patch_uses_write_limit(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(write_requests=1, window=60)
tween = rate_limit_tween_factory(handler, registry)
request = self._make_request(method="PATCH", ip="10.0.0.8")
response = tween(request)
self.assertEqual(response.status_int, 200)
request = self._make_request(method="PATCH", ip="10.0.0.8")
response = tween(request)
self.assertEqual(response.status_int, 429)
def test_delete_uses_write_limit(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(write_requests=1, window=60)
tween = rate_limit_tween_factory(handler, registry)
request = self._make_request(method="DELETE", ip="10.0.0.9")
response = tween(request)
self.assertEqual(response.status_int, 200)
request = self._make_request(method="DELETE", ip="10.0.0.9")
response = tween(request)
self.assertEqual(response.status_int, 429)
def test_retry_after_value(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry(read_requests=1, window=60)
tween = rate_limit_tween_factory(handler, registry)
request = self._make_request(ip="10.0.0.10")
tween(request)
request = self._make_request(ip="10.0.0.10")
response = tween(request)
self.assertEqual(response.status_int, 429)
retry_after = response.json_body["retry_after"]
self.assertGreater(retry_after, 0)
self.assertLessEqual(retry_after, 61)
def test_defaults_when_no_settings(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = mock.MagicMock()
registry.settings = {}
tween = rate_limit_tween_factory(handler, registry)
# Should use defaults (120 read, 30 write, 60s window)
request = self._make_request(ip="10.0.0.11")
response = tween(request)
self.assertEqual(response.status_int, 200)
def test_global_api_disabled(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry()
registry.settings["api.enabled"] = "false"
tween = rate_limit_tween_factory(handler, registry)
request = self._make_request(ip="10.0.0.12")
response = tween(request)
self.assertEqual(response.status_int, 404)
self.assertIn("disabled", response.json_body["error"])
self.assertFalse(handler.called)
def test_global_api_disabled_skips_non_api(self):
from remarkbox.api.rate_limit import rate_limit_tween_factory
handler = self._make_handler()
registry = self._make_registry()
registry.settings["api.enabled"] = "false"
tween = rate_limit_tween_factory(handler, registry)
# Non-API path should still work even if API is disabled
request = self._make_request(path="/some-page", ip="10.0.0.13")
response = tween(request)
self.assertEqual(response.status_int, 200)
self.assertTrue(handler.called)

206
remarkbox/tests/test_api_serializers.py Normal file
View file

@ -0,0 +1,206 @@
import unittest
import uuid
from unittest import mock
class TestSerializeAuthor(unittest.TestCase):
def test_with_user(self):
from remarkbox.api.serializers import serialize_author
user = mock.MagicMock()
user.id = uuid.uuid1()
user.name = "TestUser"
node = mock.MagicMock()
node.user = user
node.user_surrogate = None
result = serialize_author(node)
self.assertEqual(result["type"], "user")
self.assertEqual(result["id"], str(user.id))
self.assertEqual(result["name"], "TestUser")
def test_with_surrogate(self):
from remarkbox.api.serializers import serialize_author
surrogate = mock.MagicMock()
surrogate.id = uuid.uuid1()
surrogate.name = "AnonBot"
node = mock.MagicMock()
node.user = None
node.user_surrogate = surrogate
result = serialize_author(node)
self.assertEqual(result["type"], "surrogate")
self.assertEqual(result["id"], str(surrogate.id))
self.assertEqual(result["name"], "AnonBot")
def test_with_no_author(self):
from remarkbox.api.serializers import serialize_author
node = mock.MagicMock()
node.user = None
node.user_surrogate = None
result = serialize_author(node)
self.assertIsNone(result)
class TestSerializeNode(unittest.TestCase):
def _make_mock_node(self, **kwargs):
node = mock.MagicMock()
node.id = kwargs.get("id", uuid.uuid1())
node.root_id = kwargs.get("root_id", node.id)
node.parent_id = kwargs.get("parent_id", None)
node.title = kwargs.get("title", "Test Title")
node.data = kwargs.get("data", "Test data")
node.data_html = kwargs.get("data_html", "<p>Test data</p>")
node.is_root = kwargs.get("is_root", True)
node.graph_depth = kwargs.get("graph_depth", 0)
node.created = kwargs.get("created", 1700000000000)
node.created_date = "2023-11-14"
node.human_created_timestamp = "just now"
node.changed = kwargs.get("changed", 1700000000000)
node.changed_date = "2023-11-14"
node.human_changed_timestamp = "just now"
node.disabled = kwargs.get("disabled", False)
node.verified = kwargs.get("verified", True)
node.locked = kwargs.get("locked", False)
node.approved = kwargs.get("approved", True)
node.was_edited = kwargs.get("was_edited", False)
node.user = kwargs.get("user", None)
node.user_surrogate = kwargs.get("user_surrogate", None)
node.cache = kwargs.get("cache", None)
return node
def test_basic_fields(self):
from remarkbox.api.serializers import serialize_node
node = self._make_mock_node()
result = serialize_node(node)
self.assertEqual(result["id"], str(node.id))
self.assertEqual(result["title"], "Test Title")
self.assertEqual(result["data"], "Test data")
self.assertEqual(result["data_html"], "<p>Test data</p>")
self.assertTrue(result["is_root"])
self.assertEqual(result["depth"], 0)
self.assertFalse(result["disabled"])
self.assertTrue(result["verified"])
self.assertFalse(result["locked"])
self.assertTrue(result["approved"])
self.assertFalse(result["was_edited"])
def test_uuid_fields_are_strings(self):
from remarkbox.api.serializers import serialize_node
node_id = uuid.uuid1()
root_id = uuid.uuid1()
parent_id = uuid.uuid1()
node = self._make_mock_node(id=node_id, root_id=root_id, parent_id=parent_id)
result = serialize_node(node)
self.assertEqual(result["id"], str(node_id))
self.assertEqual(result["root_id"], str(root_id))
self.assertEqual(result["parent_id"], str(parent_id))
def test_null_parent_id(self):
from remarkbox.api.serializers import serialize_node
node = self._make_mock_node(parent_id=None)
result = serialize_node(node)
self.assertIsNone(result["parent_id"])
def test_disabled_node(self):
from remarkbox.api.serializers import serialize_node
node = self._make_mock_node(disabled=True)
result = serialize_node(node)
self.assertTrue(result["disabled"])
def test_child_node(self):
from remarkbox.api.serializers import serialize_node
root_id = uuid.uuid1()
parent_id = uuid.uuid1()
node = self._make_mock_node(
is_root=False, root_id=root_id, parent_id=parent_id, graph_depth=2
)
result = serialize_node(node)
self.assertFalse(result["is_root"])
self.assertEqual(result["depth"], 2)
self.assertEqual(result["root_id"], str(root_id))
self.assertEqual(result["parent_id"], str(parent_id))
def test_include_children_with_cache(self):
from remarkbox.api.serializers import serialize_node
cache = mock.MagicMock()
node = self._make_mock_node(cache=cache)
node.stats = {"root": {"count": 5, "visible_count": 4}}
result = serialize_node(node, include_children=True)
self.assertEqual(result["stats"], {"root": {"count": 5, "visible_count": 4}})
def test_include_children_no_cache(self):
from remarkbox.api.serializers import serialize_node
node = self._make_mock_node(cache=None)
result = serialize_node(node, include_children=True)
self.assertIsNone(result["stats"])
def test_no_stats_by_default(self):
from remarkbox.api.serializers import serialize_node
node = self._make_mock_node()
result = serialize_node(node)
self.assertNotIn("stats", result)
def test_with_user_author(self):
from remarkbox.api.serializers import serialize_node
user = mock.MagicMock()
user.id = uuid.uuid1()
user.name = "AgentSmith"
node = self._make_mock_node(user=user)
result = serialize_node(node)
self.assertEqual(result["author"]["type"], "user")
self.assertEqual(result["author"]["name"], "AgentSmith")
def test_with_surrogate_author(self):
from remarkbox.api.serializers import serialize_node
surrogate = mock.MagicMock()
surrogate.id = uuid.uuid1()
surrogate.name = "ClaudeBot"
node = self._make_mock_node(user_surrogate=surrogate)
result = serialize_node(node)
self.assertEqual(result["author"]["type"], "surrogate")
self.assertEqual(result["author"]["name"], "ClaudeBot")
def test_timestamps(self):
from remarkbox.api.serializers import serialize_node
node = self._make_mock_node(created=1700000000000, changed=1700001000000)
result = serialize_node(node)
self.assertEqual(result["created"], 1700000000000)
self.assertEqual(result["changed"], 1700001000000)
self.assertEqual(result["created_date"], "2023-11-14")
class TestSerializeNamespaceBrief(unittest.TestCase):
def test_basic_fields(self):
from remarkbox.api.serializers import serialize_namespace_brief
ns = mock.MagicMock()
ns.id = uuid.uuid1()
ns.name = "agents.example.com"
ns.description = "Agent discussion"
ns.allow_anonymous = True
ns.node_order = "oldest-first"
result = serialize_namespace_brief(ns)
self.assertEqual(result["id"], str(ns.id))
self.assertEqual(result["name"], "agents.example.com")
self.assertEqual(result["description"], "Agent discussion")
self.assertTrue(result["allow_anonymous"])
self.assertEqual(result["node_order"], "oldest-first")

851
remarkbox/tests/test_api_views.py Normal file
View file

@ -0,0 +1,851 @@
import transaction
import unittest
import webtest
from remarkbox.models import (
Node,
get_tm_session,
get_or_create_user_by_email,
get_user_by_email,
get_or_create_namespace,
UserSurrogate,
)
from remarkbox.models.meta import Base, id_to_uuid
from pyramid.paster import get_appsettings
from unittest.mock import patch
class APIFunctionalTests(unittest.TestCase, object):
"""Base class for API functional tests."""
@classmethod
def setUpClass(cls):
from remarkbox import main
cls.settings = get_appsettings("test.ini")
cls.app = main({}, **cls.settings)
cls.testapp = webtest.TestApp(cls.app)
cls.session_factory = cls.app.registry["dbsession_factory"]
cls.engine = cls.session_factory.kw["bind"]
Base.metadata.create_all(bind=cls.engine)
cls.tm = transaction.manager
cls.dbsession = get_tm_session(cls.session_factory, cls.tm)
@classmethod
def tearDownClass(cls):
cls.dbsession.close()
Base.metadata.drop_all(bind=cls.engine)
def tearDown(self):
self.testapp.get("/log-out")
class TestAPIAnonymousPosting(APIFunctionalTests):
"""Functional tests for anonymous posting via the API."""
@classmethod
def setUpClass(cls):
try:
APIFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
APIFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
ns = get_or_create_namespace(self.dbsession, "api-test.example.com")
ns.allow_anonymous = True
self.dbsession.add(ns)
self.dbsession.flush()
self.namespace_name = str(ns.name)
self.namespace_id = ns.id
self.tm.commit()
def tearDown(self):
super(TestAPIAnonymousPosting, self).tearDown()
# Clean up surrogates and nodes
self.dbsession.query(UserSurrogate).filter(
UserSurrogate.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.query(Node).filter(
Node.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.flush()
self.tm.commit()
def test_create_anonymous_thread(self):
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Agent Thread",
"data": "Hello from an AI agent",
"anonymous_name": "ClaudeBot",
},
expect_errors=True,
)
self.assertEqual(res.status_int, 201)
body = res.json
self.assertIn("node", body)
self.assertEqual(body["node"]["title"], "Agent Thread")
self.assertEqual(body["node"]["data"], "Hello from an AI agent")
self.assertIn("<p>Hello from an AI agent</p>", body["node"]["data_html"])
self.assertTrue(body["verified"])
self.assertEqual(body["node"]["author"]["type"], "surrogate")
self.assertEqual(body["node"]["author"]["name"], "ClaudeBot")
def test_create_anonymous_thread_default_name(self):
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Default Anon",
"data": "No name given",
},
expect_errors=True,
)
self.assertEqual(res.status_int, 201)
self.assertEqual(res.json["node"]["author"]["name"], "Anonymous")
def test_create_thread_missing_namespace(self):
res = self.testapp.post_json(
"/api/v1/threads",
{"title": "Test", "data": "Test"},
expect_errors=True,
)
self.assertEqual(res.status_int, 400)
self.assertIn("namespace", res.json["error"])
def test_create_thread_missing_title(self):
res = self.testapp.post_json(
"/api/v1/threads",
{"namespace": self.namespace_name, "data": "Test"},
expect_errors=True,
)
self.assertEqual(res.status_int, 400)
self.assertIn("title", res.json["error"])
def test_create_thread_missing_data(self):
res = self.testapp.post_json(
"/api/v1/threads",
{"namespace": self.namespace_name, "title": "Test"},
expect_errors=True,
)
self.assertEqual(res.status_int, 400)
self.assertIn("data", res.json["error"])
def test_create_thread_content_too_long(self):
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Test",
"data": "x" * 600000,
},
expect_errors=True,
)
self.assertEqual(res.status_int, 400)
self.assertIn("maximum length", res.json["error"])
def test_reply_to_thread(self):
# Create thread
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Thread for replies",
"data": "Original post",
"anonymous_name": "Bot1",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
# Reply
reply_res = self.testapp.post_json(
"/api/v1/threads/{}/replies".format(node_id),
{
"data": "A reply from an agent",
"anonymous_name": "Bot2",
},
expect_errors=True,
)
self.assertEqual(reply_res.status_int, 201)
self.assertIn("node", reply_res.json)
self.assertEqual(reply_res.json["node"]["author"]["name"], "Bot2")
self.assertFalse(reply_res.json["node"]["is_root"])
self.assertEqual(reply_res.json["node"]["parent_id"], node_id)
def test_reply_missing_data(self):
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Thread",
"data": "Content",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
reply_res = self.testapp.post_json(
"/api/v1/threads/{}/replies".format(node_id),
{"anonymous_name": "Bot"},
expect_errors=True,
)
self.assertEqual(reply_res.status_int, 400)
def test_reply_to_nonexistent_node(self):
res = self.testapp.post_json(
"/api/v1/threads/00000000-0000-0000-0000-000000000000/replies",
{"data": "Should fail", "anonymous_name": "Bot"},
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
def test_reply_to_locked_thread(self):
# Create thread
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Will Be Locked",
"data": "Content",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
# Lock the thread directly in DB
node = self.dbsession.query(Node).filter(
Node.id == id_to_uuid(node_id)
).first()
node.locked = True
self.dbsession.add(node)
self.dbsession.flush()
self.tm.commit()
reply_res = self.testapp.post_json(
"/api/v1/threads/{}/replies".format(node_id),
{"data": "Should fail", "anonymous_name": "Bot"},
expect_errors=True,
)
self.assertEqual(reply_res.status_int, 403)
self.assertIn("locked", reply_res.json["error"])
def test_list_threads(self):
# Create a thread
self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Listable Thread",
"data": "Some content",
"anonymous_name": "Bot",
},
expect_errors=True,
)
res = self.testapp.get(
"/api/v1/threads",
{"namespace": self.namespace_name},
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertIn("threads", res.json)
self.assertIn("namespace", res.json)
self.assertEqual(res.json["namespace"]["name"], self.namespace_name)
self.assertGreater(len(res.json["threads"]), 0)
thread = res.json["threads"][0]
self.assertIn("id", thread)
self.assertIn("title", thread)
self.assertIn("author", thread)
def test_list_threads_missing_namespace(self):
res = self.testapp.get(
"/api/v1/threads",
expect_errors=True,
)
self.assertEqual(res.status_int, 400)
def test_get_thread_detail(self):
# Create thread with a reply
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Detail Thread",
"data": "Content here",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
self.testapp.post_json(
"/api/v1/threads/{}/replies".format(node_id),
{"data": "A reply", "anonymous_name": "ReplyBot"},
expect_errors=True,
)
res = self.testapp.get(
"/api/v1/threads/{}".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertIn("thread", res.json)
self.assertIn("replies", res.json)
self.assertEqual(res.json["thread"]["id"], node_id)
self.assertEqual(len(res.json["replies"]), 1)
self.assertEqual(res.json["replies"][0]["author"]["name"], "ReplyBot")
def test_get_nonexistent_thread(self):
res = self.testapp.get(
"/api/v1/threads/00000000-0000-0000-0000-000000000000",
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
def test_get_single_node(self):
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Node Test",
"data": "Content",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
res = self.testapp.get(
"/api/v1/nodes/{}".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertEqual(res.json["node"]["id"], node_id)
def test_get_nonexistent_node(self):
res = self.testapp.get(
"/api/v1/nodes/00000000-0000-0000-0000-000000000000",
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
def test_reply_content_too_long(self):
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Thread",
"data": "Content",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
reply_res = self.testapp.post_json(
"/api/v1/threads/{}/replies".format(node_id),
{"data": "x" * 600000, "anonymous_name": "Bot"},
expect_errors=True,
)
self.assertEqual(reply_res.status_int, 400)
self.assertIn("maximum length", reply_res.json["error"])
def test_pagination(self):
res = self.testapp.get(
"/api/v1/threads",
{"namespace": self.namespace_name, "page": "1"},
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertEqual(res.json["page"], 1)
self.assertIn("page_size", res.json)
class TestAPIOTPAuthentication(APIFunctionalTests):
"""Functional tests for the OTP auth flow via API."""
@classmethod
def setUpClass(cls):
try:
APIFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
APIFunctionalTests.setUpClass.__func__(cls)
def tearDown(self):
super(TestAPIOTPAuthentication, self).tearDown()
# Clean up test users
for email in [
"api-login@example.com",
"api-verify@example.com",
"api-bad-otp@example.com",
"api-throttle@example.com",
]:
user = get_user_by_email(self.dbsession, email)
if user:
self.dbsession.delete(user)
self.dbsession.flush()
self.tm.commit()
@patch("smtplib.SMTP")
def test_login_sends_otp(self, mock_smtp):
res = self.testapp.post_json(
"/api/v1/auth/login",
{"email": "api-login@example.com"},
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertEqual(res.json["status"], "sent")
self.assertIn("api-login@example.com", res.json["message"])
def test_login_invalid_email(self):
res = self.testapp.post_json(
"/api/v1/auth/login",
{"email": "not-an-email"},
expect_errors=True,
)
self.assertEqual(res.status_int, 400)
self.assertIn("email", res.json["error"])
def test_login_missing_email(self):
res = self.testapp.post_json(
"/api/v1/auth/login",
{},
expect_errors=True,
)
self.assertEqual(res.status_int, 400)
@patch("smtplib.SMTP")
def test_login_throttle(self, mock_smtp):
# First login sends OTP
self.testapp.post_json(
"/api/v1/auth/login",
{"email": "api-throttle@example.com"},
expect_errors=True,
)
# Second login within throttle window
res = self.testapp.post_json(
"/api/v1/auth/login",
{"email": "api-throttle@example.com"},
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertEqual(res.json["status"], "throttled")
def test_verify_valid_otp(self):
user = get_or_create_user_by_email(self.dbsession, "api-verify@example.com")
raw_otp = user.new_password()
self.dbsession.add(user)
self.dbsession.flush()
self.tm.commit()
res = self.testapp.post_json(
"/api/v1/auth/verify",
{"email": "api-verify@example.com", "otp": raw_otp},
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertEqual(res.json["status"], "authenticated")
self.assertIn("user", res.json)
self.assertEqual(res.json["user"]["email"], "api-verify@example.com")
def test_verify_invalid_otp(self):
user = get_or_create_user_by_email(self.dbsession, "api-bad-otp@example.com")
user.new_password()
self.dbsession.add(user)
self.dbsession.flush()
self.tm.commit()
res = self.testapp.post_json(
"/api/v1/auth/verify",
{"email": "api-bad-otp@example.com", "otp": "000000"},
expect_errors=True,
)
self.assertEqual(res.status_int, 401)
def test_verify_missing_fields(self):
res = self.testapp.post_json(
"/api/v1/auth/verify",
{"email": "test@example.com"},
expect_errors=True,
)
self.assertEqual(res.status_int, 400)
res = self.testapp.post_json(
"/api/v1/auth/verify",
{"otp": "123456"},
expect_errors=True,
)
self.assertEqual(res.status_int, 400)
class TestAPIAuthenticatedEditing(APIFunctionalTests):
"""Functional tests for authenticated edit operations via API."""
@classmethod
def setUpClass(cls):
try:
APIFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
APIFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
self.test_user = get_or_create_user_by_email(
self.dbsession, "api-edit@remarkbox.com"
)
self.raw_otp = self.test_user.new_password()
self.dbsession.add(self.test_user)
self.dbsession.flush()
self.tm.commit()
self.test_user = get_or_create_user_by_email(
self.dbsession, "api-edit@remarkbox.com"
)
def tearDown(self):
super(TestAPIAuthenticatedEditing, self).tearDown()
# Clean up nodes created by this user
self.dbsession.query(Node).filter(
Node.user_id == self.test_user.id
).delete(synchronize_session=False)
self.dbsession.delete(self.test_user)
self.dbsession.flush()
self.tm.commit()
def _login(self):
self.testapp.post(
"/verification-challenge?email={}&raw-otp={}".format(
"api-edit@remarkbox.com", self.raw_otp
)
)
def test_edit_own_node(self):
self._login()
# Create thread as authenticated user
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": "localhost",
"title": "Editable Thread",
"data": "Original content",
},
expect_errors=True,
)
self.assertEqual(create_res.status_int, 201)
node_id = create_res.json["node"]["id"]
# Edit it
edit_res = self.testapp.patch_json(
"/api/v1/nodes/{}".format(node_id),
{"data": "Edited content"},
expect_errors=True,
)
self.assertEqual(edit_res.status_int, 200)
self.assertEqual(edit_res.json["node"]["data"], "Edited content")
def test_edit_title_on_root(self):
self._login()
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": "localhost",
"title": "Original Title",
"data": "Content",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
edit_res = self.testapp.patch_json(
"/api/v1/nodes/{}".format(node_id),
{"title": "New Title"},
expect_errors=True,
)
self.assertEqual(edit_res.status_int, 200)
self.assertEqual(edit_res.json["node"]["title"], "New Title")
def test_edit_requires_auth(self):
self._login()
# Create a node, then log out and try to edit
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": "localhost",
"title": "Auth Test Thread",
"data": "Content",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
self.testapp.get("/log-out")
res = self.testapp.patch_json(
"/api/v1/nodes/{}".format(node_id),
{"data": "Should fail"},
expect_errors=True,
)
self.assertEqual(res.status_int, 401)
def test_edit_nonexistent_node(self):
self._login()
res = self.testapp.patch_json(
"/api/v1/nodes/00000000-0000-0000-0000-000000000000",
{"data": "Should fail"},
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
def test_edit_missing_data_and_title(self):
self._login()
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": "localhost",
"title": "Thread",
"data": "Content",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
edit_res = self.testapp.patch_json(
"/api/v1/nodes/{}".format(node_id),
{},
expect_errors=True,
)
self.assertEqual(edit_res.status_int, 400)
def test_edit_content_too_long(self):
self._login()
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": "localhost",
"title": "Thread",
"data": "Content",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
edit_res = self.testapp.patch_json(
"/api/v1/nodes/{}".format(node_id),
{"data": "x" * 600000},
expect_errors=True,
)
self.assertEqual(edit_res.status_int, 400)
def test_authenticated_thread_creation(self):
self._login()
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": "localhost",
"title": "Auth Thread",
"data": "Authenticated post",
},
expect_errors=True,
)
self.assertEqual(res.status_int, 201)
self.assertTrue(res.json["verified"])
self.assertEqual(res.json["node"]["author"]["type"], "user")
class TestAPIDisabledNode(APIFunctionalTests):
"""Test replying to disabled nodes."""
@classmethod
def setUpClass(cls):
try:
APIFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
APIFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
ns = get_or_create_namespace(self.dbsession, "api-disabled.example.com")
ns.allow_anonymous = True
self.dbsession.add(ns)
self.dbsession.flush()
self.namespace_name = str(ns.name)
self.namespace_id = ns.id
self.tm.commit()
def tearDown(self):
super(TestAPIDisabledNode, self).tearDown()
self.dbsession.query(UserSurrogate).filter(
UserSurrogate.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.query(Node).filter(
Node.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.flush()
self.tm.commit()
def test_reply_to_disabled_node(self):
# Create thread
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Will Be Disabled",
"data": "Content",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
# Disable it
node = self.dbsession.query(Node).filter(
Node.id == id_to_uuid(node_id)
).first()
node.disabled = True
self.dbsession.add(node)
self.dbsession.flush()
self.tm.commit()
# Try to reply
reply_res = self.testapp.post_json(
"/api/v1/threads/{}/replies".format(node_id),
{"data": "Should fail", "anonymous_name": "Bot"},
expect_errors=True,
)
self.assertEqual(reply_res.status_int, 403)
self.assertIn("disabled", reply_res.json["error"])
class TestAPINamespaceOptOut(APIFunctionalTests):
"""Test per-namespace api_access opt-out."""
@classmethod
def setUpClass(cls):
try:
APIFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
APIFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
ns = get_or_create_namespace(self.dbsession, "api-blocked.example.com")
ns.allow_anonymous = True
ns.api_access = False
self.dbsession.add(ns)
self.dbsession.flush()
self.namespace_name = str(ns.name)
self.namespace_id = ns.id
self.tm.commit()
def tearDown(self):
super(TestAPINamespaceOptOut, self).tearDown()
self.dbsession.query(UserSurrogate).filter(
UserSurrogate.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.query(Node).filter(
Node.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.flush()
self.tm.commit()
def test_list_threads_blocked(self):
res = self.testapp.get(
"/api/v1/threads",
{"namespace": self.namespace_name},
expect_errors=True,
)
self.assertEqual(res.status_int, 403)
self.assertIn("disabled", res.json["error"])
def test_create_thread_blocked(self):
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Should fail",
"data": "Content",
"anonymous_name": "Bot",
},
expect_errors=True,
)
self.assertEqual(res.status_int, 403)
def test_reply_blocked(self):
# Re-enable temporarily to create a thread, then disable
ns = get_or_create_namespace(self.dbsession, self.namespace_name)
ns.api_access = True
self.dbsession.add(ns)
self.dbsession.flush()
self.tm.commit()
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Thread",
"data": "Content",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
# Disable API access
ns = get_or_create_namespace(self.dbsession, self.namespace_name)
ns.api_access = False
self.dbsession.add(ns)
self.dbsession.flush()
self.tm.commit()
reply_res = self.testapp.post_json(
"/api/v1/threads/{}/replies".format(node_id),
{"data": "Should fail", "anonymous_name": "Bot"},
expect_errors=True,
)
self.assertEqual(reply_res.status_int, 403)
def test_get_thread_blocked(self):
# Re-enable temporarily to create a thread, then disable
ns = get_or_create_namespace(self.dbsession, self.namespace_name)
ns.api_access = True
self.dbsession.add(ns)
self.dbsession.flush()
self.tm.commit()
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Thread",
"data": "Content",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
# Disable API access
ns = get_or_create_namespace(self.dbsession, self.namespace_name)
ns.api_access = False
self.dbsession.add(ns)
self.dbsession.flush()
self.tm.commit()
res = self.testapp.get(
"/api/v1/threads/{}".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 403)

11
remarkbox/views/authenticated/authenticated.py
View file

@ -70,6 +70,7 @@ def namespace_settings(request):
mathjax_checkbox = p.get("mathjax-checkbox", "off")
ignore_query_string_checkbox = p.get("ignore-query-string-checkbox", "off")
hide_powered_by_checkbox = p.get("hide-powered-by-checkbox", "off")
api_access_checkbox = p.get("api-access-checkbox", "off")
hide_unless_approved = checkbox_to_bool(hide_unless_approved_checkbox)
allow_anonymous = checkbox_to_bool(allow_anonymous_checkbox)
@ -79,6 +80,7 @@ def namespace_settings(request):
mathjax = checkbox_to_bool(mathjax_checkbox)
ignore_query_string = checkbox_to_bool(ignore_query_string_checkbox)
hide_powered_by = checkbox_to_bool(hide_powered_by_checkbox)
api_access = checkbox_to_bool(api_access_checkbox)
if stylesheet_embed != request.namespace.stylesheet_embed and (
stylesheet_embed or request.namespace.stylesheet_embed
@ -228,6 +230,15 @@ def namespace_settings(request):
("You turned {} MathJax".format(mathjax_checkbox), "success")
)
if api_access != request.namespace.api_access:
request.namespace.api_access = api_access
request.session.flash(
(
"You turned {} api_access".format(api_access_checkbox),
"success",
)
)
request.dbsession.add(request.namespace)
request.dbsession.flush()

8
test.ini
View file

@ -35,6 +35,14 @@ session.timeout = 31104000
session.max_age = 31104000
session.reissue_time = 15552000
###
# API configuration (permissive for tests).
###
api.enabled = true
api.rate_limit.read_requests = 1000
api.rate_limit.write_requests = 1000
api.rate_limit.window = 60
###
# custom app configuration.
###