'''
def get_user_from_cookie(request: Request) -> Optional[str]:
"""Get username from auth cookie."""
token = request.cookies.get("auth_token")
if token:
return verify_token(token)
return None
def wants_html(request: Request) -> bool:
"""Check if request wants HTML (browser) vs JSON (API)."""
accept = request.headers.get("accept", "")
return "text/html" in accept and "application/json" not in accept and "application/activity+json" not in accept
def format_date(value, length: int = 10) -> str:
"""Format a date value (datetime or string) to a string, sliced to length."""
if value is None:
return ""
if hasattr(value, 'isoformat'):
return value.isoformat()[:length]
if isinstance(value, str):
return value[:length]
return ""
# ============ Auth UI Endpoints ============
@app.get("/login", response_class=HTMLResponse)
async def ui_login_page(request: Request, return_to: str = None):
"""Login page. Accepts optional return_to URL for redirect after login."""
username = get_user_from_cookie(request)
if username:
return HTMLResponse(base_html("Already Logged In", f'''
'''
return HTMLResponse(base_html("Login", content))
@app.post("/login", response_class=HTMLResponse)
async def ui_login_submit(request: Request):
"""Handle login form submission."""
form = await request.form()
username = form.get("username", "").strip()
password = form.get("password", "")
return_to = form.get("return_to", "").strip()
if not username or not password:
return HTMLResponse('
Username and password are required
')
user = await authenticate_user(DATA_DIR, username, password)
if not user:
return HTMLResponse('
Invalid username or password
')
token = create_access_token(user.username, l2_server=f"https://{DOMAIN}")
# If return_to is specified, redirect there with token for the other site to set its own cookie
if return_to and return_to.startswith("http"):
# Append token to return_to URL for the target site to set its own cookie
separator = "&" if "?" in return_to else "?"
redirect_url = f"{return_to}{separator}auth_token={token.access_token}"
response = HTMLResponse(f'''
Login successful! Redirecting...
''')
else:
response = HTMLResponse(f'''
Login successful! Redirecting...
''')
# Set cookie for L2 only (L1 servers set their own cookies via /auth endpoint)
response.set_cookie(
key="auth_token",
value=token.access_token,
httponly=True,
max_age=60 * 60 * 24 * 30, # 30 days
samesite="lax",
secure=True
)
return response
@app.get("/register", response_class=HTMLResponse)
async def ui_register_page(request: Request):
"""Register page."""
username = get_user_from_cookie(request)
if username:
return HTMLResponse(base_html("Already Logged In", f'''
''')
response.set_cookie(
key="auth_token",
value=token.access_token,
httponly=True,
max_age=60 * 60 * 24 * 30, # 30 days
samesite="lax",
secure=True
)
return response
@app.get("/logout")
async def logout(request: Request):
"""Handle logout - clear cookie, revoke token on L2 and attached L1s, and redirect to home."""
token = request.cookies.get("auth_token")
claims = get_token_claims(token) if token else None
username = claims.get("sub") if claims else None
if username and token and claims:
# Revoke token in L2 database (so even if L1 ignores revoke, token won't verify)
token_hash = hashlib.sha256(token.encode()).hexdigest()
expires_at = datetime.fromtimestamp(claims.get("exp", 0), tz=timezone.utc)
await db.revoke_token(token_hash, username, expires_at)
# Revoke ALL tokens for this user on attached L1 renderers
# (L1 may have scoped tokens different from L2's token)
attached = await db.get_user_renderers(username)
for l1_url in attached:
try:
requests.post(
f"{l1_url}/auth/revoke-user",
json={"username": username, "l2_server": f"https://{DOMAIN}"},
timeout=5
)
except Exception as e:
logger.warning(f"Failed to revoke user tokens on {l1_url}: {e}")
# Remove all attachments for this user
for l1_url in attached:
await db.detach_renderer(username, l1_url)
response = RedirectResponse(url="/", status_code=302)
# Delete both legacy (no domain) and new (shared domain) cookies
response.delete_cookie("auth_token")
if COOKIE_DOMAIN:
response.delete_cookie("auth_token", domain=COOKIE_DOMAIN)
return response
# ============ HTML Rendering Helpers ============
async def ui_activity_detail(activity_index: int, request: Request):
"""Activity detail page with full content display. Helper function for HTML rendering."""
username = get_user_from_cookie(request)
activities = await load_activities()
if activity_index < 0 or activity_index >= len(activities):
content = '''
'''
return HTMLResponse(base_html(f"Activity: {obj_name}", content, username))
async def ui_asset_detail(name: str, request: Request):
"""Asset detail page with content preview and provenance. Helper function for HTML rendering."""
username = get_user_from_cookie(request)
registry = await load_registry()
assets = registry.get("assets", {})
if name not in assets:
content = f'''
'''
return HTMLResponse(base_html(f"Asset: {name}", content, username))
async def ui_user_detail(username: str, request: Request):
"""User detail page showing their published assets. Helper function for HTML rendering."""
current_user = get_user_from_cookie(request)
if not await user_exists(username):
content = f'''
'''
# For infinite scroll, just return rows if not first page
if page > 1:
if has_more:
rows += f'''
Loading more...
'''
return HTMLResponse(rows)
# First page - full content
infinite_scroll_trigger = ""
if has_more:
infinite_scroll_trigger = f'''
Loading more...
'''
content = f'''
Users ({total} total)
Username
WebFinger
Created
{rows}
{infinite_scroll_trigger}
'''
return HTMLResponse(base_html("Users", content, username))
# JSON response for APIs
return {
"users": [{"username": uname, **data} for uname, data in users_page],
"pagination": {
"page": page,
"limit": limit,
"total": total,
"has_more": has_more
}
}
@app.get("/users/{username}")
async def get_actor(username: str, request: Request):
"""Get actor profile for any registered user. Content negotiation: HTML for browsers, JSON for APIs."""
if not await user_exists(username):
if wants_html(request):
content = f'''
'''
# For infinite scroll, just return rows if not first page
if page > 1:
if has_more:
rows += f'''
Loading more...
'''
return HTMLResponse(rows)
# First page - full content with header
infinite_scroll_trigger = ""
if has_more:
infinite_scroll_trigger = f'''
Loading more...
'''
content = f'''
Activities ({total} total)
Type
Object
Actor
Published
{rows}
{infinite_scroll_trigger}
'''
return HTMLResponse(base_html("Activities", content, username))
# JSON response for APIs
return {
"activities": activities_page,
"pagination": {
"page": page,
"limit": limit,
"total": total,
"has_more": has_more
}
}
@app.get("/activities/{activity_ref}")
async def get_activity_detail(activity_ref: str, request: Request):
"""Get single activity by index or activity_id. HTML for browsers (default), JSON only if explicitly requested."""
# Check if JSON explicitly requested
accept = request.headers.get("accept", "")
wants_json = ("application/json" in accept or "application/activity+json" in accept) and "text/html" not in accept
activity = None
activity_index = None
# Check if it's a numeric index or an activity_id (hash)
if activity_ref.isdigit():
# Numeric index (legacy)
activity_index = int(activity_ref)
activities = await load_activities()
if 0 <= activity_index < len(activities):
activity = activities[activity_index]
else:
# Activity ID (hash) - look up directly
activity = await db.get_activity(activity_ref)
if activity:
# Find index for UI rendering
activities = await load_activities()
for i, a in enumerate(activities):
if a.get("activity_id") == activity_ref:
activity_index = i
break
if not activity:
if wants_json:
raise HTTPException(404, "Activity not found")
content = '''
'''
return HTMLResponse(base_html("Activity Not Found", content, get_user_from_cookie(request)))
if wants_json:
return activity
# Default to HTML for browsers
if activity_index is not None:
return await ui_activity_detail(activity_index, request)
else:
# Render activity directly if no index found
return await ui_activity_detail_by_data(activity, request)
@app.get("/activity/{activity_index}")
async def get_activity_legacy(activity_index: int):
"""Legacy route - redirect to /activities/{activity_index}."""
return RedirectResponse(url=f"/activities/{activity_index}", status_code=301)
@app.get("/objects/{content_hash}")
async def get_object(content_hash: str, request: Request):
"""Get object by content hash. Content negotiation: HTML for browsers, JSON for APIs."""
registry = await load_registry()
# Find asset by hash
for name, asset in registry.get("assets", {}).items():
if asset.get("content_hash") == content_hash:
# Check Accept header - only return JSON if explicitly requested
accept = request.headers.get("accept", "")
wants_json = ("application/json" in accept or "application/activity+json" in accept) and "text/html" not in accept
if not wants_json:
# Default: redirect to detail page for browsers
return RedirectResponse(url=f"/assets/{name}", status_code=303)
owner = asset.get("owner", "unknown")
return JSONResponse(
content={
"@context": "https://www.w3.org/ns/activitystreams",
"id": f"https://{DOMAIN}/objects/{content_hash}",
"type": asset.get("asset_type", "Object").capitalize(),
"name": name,
"contentHash": {
"algorithm": "sha3-256",
"value": content_hash
},
"attributedTo": f"https://{DOMAIN}/users/{owner}",
"published": asset.get("created_at")
},
media_type="application/activity+json"
)
raise HTTPException(404, f"Object not found: {content_hash}")
# ============ Anchoring (Bitcoin timestamps) ============
@app.post("/anchors/create")
async def create_anchor_endpoint(request: Request):
"""
Create a new anchor for all unanchored activities.
Builds a merkle tree, stores it on IPFS, and submits to OpenTimestamps
for Bitcoin anchoring. The anchor proof is backed up to persistent storage.
"""
import anchoring
import ipfs_client
# Check auth (cookie or header)
username = get_user_from_cookie(request)
if not username:
if wants_html(request):
return HTMLResponse('''
Error: Login required
''')
raise HTTPException(401, "Authentication required")
# Get unanchored activities
unanchored = await db.get_unanchored_activities()
if not unanchored:
if wants_html(request):
return HTMLResponse('''
No unanchored activities to anchor.
''')
return {"message": "No unanchored activities", "anchored": 0}
activity_ids = [a["activity_id"] for a in unanchored]
# Create anchor
anchor = await anchoring.create_anchor(activity_ids, db, ipfs_client)
if anchor:
if wants_html(request):
return HTMLResponse(f'''
Success! Anchored {len(activity_ids)} activities. Merkle root: {anchor["merkle_root"][:32]}... Refresh page to see the new anchor.
''')
raise HTTPException(500, "Failed to create anchor")
@app.get("/anchors")
async def list_anchors():
"""List all anchors."""
anchors = await db.get_all_anchors()
stats = await db.get_anchor_stats()
return {
"anchors": anchors,
"stats": stats
}
@app.get("/anchors/{merkle_root}")
async def get_anchor_endpoint(merkle_root: str):
"""Get anchor details by merkle root."""
anchor = await db.get_anchor(merkle_root)
if not anchor:
raise HTTPException(404, f"Anchor not found: {merkle_root}")
return anchor
@app.get("/anchors/{merkle_root}/tree")
async def get_anchor_tree(merkle_root: str):
"""Get the full merkle tree from IPFS."""
import asyncio
import ipfs_client
anchor = await db.get_anchor(merkle_root)
if not anchor:
raise HTTPException(404, f"Anchor not found: {merkle_root}")
tree_cid = anchor.get("tree_ipfs_cid")
if not tree_cid:
raise HTTPException(404, "Anchor has no tree on IPFS")
try:
tree_bytes = await asyncio.to_thread(ipfs_client.get_bytes, tree_cid)
if tree_bytes:
return json.loads(tree_bytes)
except Exception as e:
raise HTTPException(500, f"Failed to fetch tree from IPFS: {e}")
@app.get("/anchors/verify/{activity_id}")
async def verify_activity_anchor(activity_id: str):
"""
Verify an activity's anchor proof.
Returns the merkle proof showing this activity is included in an anchored batch.
"""
import anchoring
import ipfs_client
# Get activity
activity = await db.get_activity(activity_id)
if not activity:
raise HTTPException(404, f"Activity not found: {activity_id}")
anchor_root = activity.get("anchor_root")
if not anchor_root:
return {"verified": False, "reason": "Activity not yet anchored"}
# Get anchor
anchor = await db.get_anchor(anchor_root)
if not anchor:
return {"verified": False, "reason": "Anchor record not found"}
# Get tree from IPFS (non-blocking)
import asyncio
tree_cid = anchor.get("tree_ipfs_cid")
if not tree_cid:
return {"verified": False, "reason": "Merkle tree not on IPFS"}
try:
tree_bytes = await asyncio.to_thread(ipfs_client.get_bytes, tree_cid)
tree = json.loads(tree_bytes) if tree_bytes else None
except Exception:
return {"verified": False, "reason": "Failed to fetch tree from IPFS"}
if not tree:
return {"verified": False, "reason": "Could not load merkle tree"}
# Get proof
proof = anchoring.get_merkle_proof(tree, activity_id)
if not proof:
return {"verified": False, "reason": "Activity not in merkle tree"}
# Verify proof
valid = anchoring.verify_merkle_proof(activity_id, proof, anchor_root)
return {
"verified": valid,
"activity_id": activity_id,
"merkle_root": anchor_root,
"tree_ipfs_cid": tree_cid,
"proof": proof,
"bitcoin_txid": anchor.get("bitcoin_txid"),
"confirmed_at": anchor.get("confirmed_at")
}
@app.post("/anchors/{merkle_root}/upgrade")
async def upgrade_anchor_proof(merkle_root: str):
"""
Try to upgrade an OTS proof from pending to confirmed.
Bitcoin confirmation typically takes 1-2 hours. Call this periodically
to check if the proof has been included in a Bitcoin block.
"""
import anchoring
import ipfs_client
import asyncio
anchor = await db.get_anchor(merkle_root)
if not anchor:
raise HTTPException(404, f"Anchor not found: {merkle_root}")
if anchor.get("confirmed_at"):
return {"status": "already_confirmed", "bitcoin_txid": anchor.get("bitcoin_txid")}
# Get current OTS proof from IPFS
ots_cid = anchor.get("ots_proof_cid")
if not ots_cid:
return {"status": "no_proof", "message": "No OTS proof stored"}
try:
ots_proof = await asyncio.to_thread(ipfs_client.get_bytes, ots_cid)
if not ots_proof:
return {"status": "error", "message": "Could not fetch OTS proof from IPFS"}
except Exception as e:
return {"status": "error", "message": f"IPFS error: {e}"}
# Try to upgrade
upgraded = await asyncio.to_thread(anchoring.upgrade_ots_proof, ots_proof)
if upgraded and len(upgraded) > len(ots_proof):
# Store upgraded proof on IPFS
try:
new_cid = await asyncio.to_thread(ipfs_client.add_bytes, upgraded)
# TODO: Update anchor record with new CID and confirmed status
return {
"status": "upgraded",
"message": "Proof upgraded - Bitcoin confirmation received",
"new_ots_cid": new_cid,
"proof_size": len(upgraded)
}
except Exception as e:
return {"status": "error", "message": f"Failed to store upgraded proof: {e}"}
else:
return {
"status": "pending",
"message": "Not yet confirmed on Bitcoin. Try again in ~1 hour.",
"proof_size": len(ots_proof) if ots_proof else 0
}
@app.get("/anchors/ui", response_class=HTMLResponse)
async def anchors_ui(request: Request):
"""Anchors UI page - view and test OpenTimestamps anchoring."""
username = get_user_from_cookie(request)
anchors = await db.get_all_anchors()
stats = await db.get_anchor_stats()
# Build anchors table rows
rows = ""
for anchor in anchors:
status = "confirmed" if anchor.get("confirmed_at") else "pending"
status_class = "text-green-400" if status == "confirmed" else "text-yellow-400"
merkle_root = anchor.get("merkle_root", "")[:16] + "..."
rows += f'''
{merkle_root}
{anchor.get("activity_count", 0)}
{status}
{format_date(anchor.get("created_at"), 16)}
'''
if not rows:
rows = '
No anchors yet
'
content = f'''
Bitcoin Anchoring via OpenTimestamps
{stats.get("total_anchors", 0)}
Total Anchors
{stats.get("confirmed_anchors", 0)}
Confirmed
{stats.get("pending_anchors", 0)}
Pending
Test Anchoring
Create a test anchor for unanchored activities, or test the OTS connection.
Anchors
Merkle Root
Activities
Status
Created
Actions
{rows}
How it works:
Activities are batched and hashed into a merkle tree
The merkle root is submitted to OpenTimestamps
OTS aggregates hashes and anchors to Bitcoin (~1-2 hours)
Once confirmed, anyone can verify the timestamp
'''
return HTMLResponse(base_html("Anchors", content, username))
@app.post("/anchors/test-ots", response_class=HTMLResponse)
async def test_ots_connection():
"""Test OpenTimestamps connection by submitting a test hash."""
import anchoring
import hashlib
import asyncio
# Create a test hash
test_data = f"test-{datetime.now(timezone.utc).isoformat()}"
test_hash = hashlib.sha256(test_data.encode()).hexdigest()
# Try to submit
try:
ots_proof = await asyncio.to_thread(anchoring.submit_to_opentimestamps, test_hash)
if ots_proof:
return HTMLResponse(f'''
Success! OpenTimestamps is working. Test hash: {test_hash[:32]}... Proof size: {len(ots_proof)} bytes
''')
else:
return HTMLResponse('''
Failed! Could not reach OpenTimestamps servers.
''')
except Exception as e:
return HTMLResponse(f'''
Error: {str(e)}
''')
# ============ Renderers (L1 servers) ============
@app.get("/renderers", response_class=HTMLResponse)
async def renderers_page(request: Request):
"""Page to manage L1 renderer attachments."""
username = get_user_from_cookie(request)
if not username:
content = '''
Renderers
Log in to manage your renderer connections.
'''
return HTMLResponse(base_html("Renderers", content))
# Get user's attached renderers
attached = await db.get_user_renderers(username)
from urllib.parse import quote
# Build renderer list
rows = []
for l1_url in L1_SERVERS:
is_attached = l1_url in attached
# Extract display name from URL
display_name = l1_url.replace("https://", "").replace("http://", "")
if is_attached:
status = 'Attached'
action = f'''
Open
'''
else:
status = 'Not attached'
# Attach via endpoint that creates scoped token (not raw token in URL)
attach_url = f"/renderers/attach?l1_url={quote(l1_url, safe='')}"
action = f'''
Attach
'''
row_id = l1_url.replace("://", "-").replace("/", "-").replace(".", "-")
rows.append(f'''
{display_name}
{l1_url}
{status}
{action}
''')
content = f'''
Renderers
Connect to L1 rendering servers. After attaching, you can run effects and manage media on that renderer.
{"".join(rows) if rows else '
No renderers configured.
'}
'''
return HTMLResponse(base_html("Renderers", content, username))
@app.get("/renderers/attach")
async def attach_renderer_redirect(request: Request, l1_url: str):
"""Create a scoped token and redirect to L1 for attachment."""
username = get_user_from_cookie(request)
if not username:
return RedirectResponse(url="/login", status_code=302)
# Verify L1 is in our allowed list
l1_normalized = l1_url.rstrip("/")
if not any(l1_normalized == s.rstrip("/") for s in L1_SERVERS):
raise HTTPException(403, f"L1 server not authorized: {l1_url}")
# Create a scoped token that only works for this specific L1
scoped_token = create_access_token(
username,
l2_server=f"https://{DOMAIN}",
l1_server=l1_normalized
)
# Redirect to L1 with scoped token
redirect_url = f"{l1_normalized}/auth?auth_token={scoped_token.access_token}"
return RedirectResponse(url=redirect_url, status_code=302)
@app.post("/renderers/detach", response_class=HTMLResponse)
async def detach_renderer(request: Request):
"""Detach from an L1 renderer."""
username = get_user_from_cookie(request)
if not username:
return HTMLResponse('
Not logged in
')
form = await request.form()
l1_url = form.get("l1_url", "")
await db.detach_renderer(username, l1_url)
# Return updated row with link to attach endpoint (not raw token)
display_name = l1_url.replace("https://", "").replace("http://", "")
from urllib.parse import quote
attach_url = f"/renderers/attach?l1_url={quote(l1_url, safe='')}"
row_id = l1_url.replace("://", "-").replace("/", "-").replace(".", "-")
return HTMLResponse(f'''
''')
# ============ User Storage ============
import storage_providers
@app.get("/storage")
async def list_storage(request: Request, user: User = Depends(get_optional_user)):
"""List user's storage providers. HTML for browsers (default), JSON only if explicitly requested."""
# Check if JSON explicitly requested
accept = request.headers.get("accept", "")
wants_json = "application/json" in accept and "text/html" not in accept
# For browser sessions, also check cookie authentication
username = user.username if user else get_user_from_cookie(request)
if not username:
if wants_json:
raise HTTPException(401, "Authentication required")
return RedirectResponse(url="/login", status_code=302)
storages = await db.get_user_storage(username)
# Add usage stats to each storage
for storage in storages:
usage = await db.get_storage_usage(storage["id"])
storage["used_bytes"] = usage["used_bytes"]
storage["pin_count"] = usage["pin_count"]
storage["donated_gb"] = storage["capacity_gb"] // 2
# Mask sensitive config keys for display
if storage.get("config"):
config = storage["config"] if isinstance(storage["config"], dict) else json.loads(storage["config"])
masked = {}
for k, v in config.items():
if "key" in k.lower() or "token" in k.lower() or "secret" in k.lower():
masked[k] = v[:4] + "..." + v[-4:] if len(str(v)) > 8 else "****"
else:
masked[k] = v
storage["config_display"] = masked
if wants_json:
return {"storages": storages}
# Default to HTML for browsers
return await ui_storage_page(username, storages, request)
@app.post("/storage")
async def add_storage(req: AddStorageRequest, user: User = Depends(get_required_user)):
"""Add a storage provider."""
# Validate provider type
valid_types = ["pinata", "web3storage", "nftstorage", "infura", "filebase", "storj", "local"]
if req.provider_type not in valid_types:
raise HTTPException(400, f"Invalid provider type: {req.provider_type}")
# Test the provider connection before saving
provider = storage_providers.create_provider(req.provider_type, {
**req.config,
"capacity_gb": req.capacity_gb
})
if not provider:
raise HTTPException(400, "Failed to create provider with given config")
success, message = await provider.test_connection()
if not success:
raise HTTPException(400, f"Provider connection failed: {message}")
# Save to database
provider_name = req.provider_name or f"{req.provider_type}-{user.username}"
storage_id = await db.add_user_storage(
username=user.username,
provider_type=req.provider_type,
provider_name=provider_name,
config=req.config,
capacity_gb=req.capacity_gb
)
if not storage_id:
raise HTTPException(500, "Failed to save storage provider")
return {"id": storage_id, "message": f"Storage provider added: {provider_name}"}
@app.post("/storage/add")
async def add_storage_form(
request: Request,
provider_type: str = Form(...),
provider_name: Optional[str] = Form(None),
description: Optional[str] = Form(None),
capacity_gb: int = Form(5),
api_key: Optional[str] = Form(None),
secret_key: Optional[str] = Form(None),
api_token: Optional[str] = Form(None),
project_id: Optional[str] = Form(None),
project_secret: Optional[str] = Form(None),
access_key: Optional[str] = Form(None),
bucket: Optional[str] = Form(None),
path: Optional[str] = Form(None),
):
"""Add a storage provider via HTML form (cookie auth)."""
username = get_user_from_cookie(request)
if not username:
return HTMLResponse('
Not authenticated
', status_code=401)
# Validate provider type
valid_types = ["pinata", "web3storage", "nftstorage", "infura", "filebase", "storj", "local"]
if provider_type not in valid_types:
return HTMLResponse(f'
Invalid provider type: {provider_type}
')
# Build config based on provider type
config = {}
if provider_type == "pinata":
if not api_key or not secret_key:
return HTMLResponse('
Pinata requires API Key and Secret Key
')
config = {"api_key": api_key, "secret_key": secret_key}
elif provider_type == "web3storage":
if not api_token:
return HTMLResponse('
web3.storage requires API Token
')
config = {"api_token": api_token}
elif provider_type == "nftstorage":
if not api_token:
return HTMLResponse('
NFT.Storage requires API Token
')
config = {"api_token": api_token}
elif provider_type == "infura":
if not project_id or not project_secret:
return HTMLResponse('
Infura requires Project ID and Project Secret
')
config = {"project_id": project_id, "project_secret": project_secret}
elif provider_type == "filebase":
if not access_key or not secret_key or not bucket:
return HTMLResponse('
Filebase requires Access Key, Secret Key, and Bucket
')
config = {"access_key": access_key, "secret_key": secret_key, "bucket": bucket}
elif provider_type == "storj":
if not access_key or not secret_key or not bucket:
return HTMLResponse('
Storj requires Access Key, Secret Key, and Bucket
')
config = {"access_key": access_key, "secret_key": secret_key, "bucket": bucket}
elif provider_type == "local":
if not path:
return HTMLResponse('
Local storage requires a path
')
config = {"path": path}
# Test the provider connection before saving
provider = storage_providers.create_provider(provider_type, {
**config,
"capacity_gb": capacity_gb
})
if not provider:
return HTMLResponse('
Failed to create provider with given config
')
success, message = await provider.test_connection()
if not success:
return HTMLResponse(f'
Provider connection failed: {message}
')
# Save to database
name = provider_name or f"{provider_type}-{username}-{len(await db.get_user_storage_by_type(username, provider_type)) + 1}"
storage_id = await db.add_user_storage(
username=username,
provider_type=provider_type,
provider_name=name,
config=config,
capacity_gb=capacity_gb,
description=description
)
if not storage_id:
return HTMLResponse('
Failed to save storage provider
')
return HTMLResponse(f'''
Storage provider "{name}" added successfully!
''')
@app.get("/storage/{storage_id}")
async def get_storage(storage_id: int, user: User = Depends(get_required_user)):
"""Get a specific storage provider."""
storage = await db.get_storage_by_id(storage_id)
if not storage:
raise HTTPException(404, "Storage provider not found")
if storage["username"] != user.username:
raise HTTPException(403, "Not authorized")
usage = await db.get_storage_usage(storage_id)
storage["used_bytes"] = usage["used_bytes"]
storage["pin_count"] = usage["pin_count"]
storage["donated_gb"] = storage["capacity_gb"] // 2
return storage
@app.patch("/storage/{storage_id}")
async def update_storage(storage_id: int, req: UpdateStorageRequest, user: User = Depends(get_required_user)):
"""Update a storage provider."""
storage = await db.get_storage_by_id(storage_id)
if not storage:
raise HTTPException(404, "Storage provider not found")
if storage["username"] != user.username:
raise HTTPException(403, "Not authorized")
# If updating config, test the new connection
if req.config:
existing_config = storage["config"] if isinstance(storage["config"], dict) else json.loads(storage["config"])
new_config = {**existing_config, **req.config}
provider = storage_providers.create_provider(storage["provider_type"], {
**new_config,
"capacity_gb": req.capacity_gb or storage["capacity_gb"]
})
if provider:
success, message = await provider.test_connection()
if not success:
raise HTTPException(400, f"Provider connection failed: {message}")
success = await db.update_user_storage(
storage_id,
config=req.config,
capacity_gb=req.capacity_gb,
is_active=req.is_active
)
if not success:
raise HTTPException(500, "Failed to update storage provider")
return {"message": "Storage provider updated"}
@app.delete("/storage/{storage_id}")
async def remove_storage(storage_id: int, request: Request, user: User = Depends(get_optional_user)):
"""Remove a storage provider."""
# Support both Bearer token and cookie auth
username = user.username if user else get_user_from_cookie(request)
if not username:
raise HTTPException(401, "Not authenticated")
storage = await db.get_storage_by_id(storage_id)
if not storage:
raise HTTPException(404, "Storage provider not found")
if storage["username"] != username:
raise HTTPException(403, "Not authorized")
success = await db.remove_user_storage(storage_id)
if not success:
raise HTTPException(500, "Failed to remove storage provider")
# Return empty string for HTMX to remove the element
if wants_html(request):
return HTMLResponse("")
return {"message": "Storage provider removed"}
@app.post("/storage/{storage_id}/test")
async def test_storage(storage_id: int, request: Request, user: User = Depends(get_optional_user)):
"""Test storage provider connectivity."""
# Support both Bearer token and cookie auth
username = user.username if user else get_user_from_cookie(request)
if not username:
if wants_html(request):
return HTMLResponse('Not authenticated', status_code=401)
raise HTTPException(401, "Not authenticated")
storage = await db.get_storage_by_id(storage_id)
if not storage:
if wants_html(request):
return HTMLResponse('Storage not found', status_code=404)
raise HTTPException(404, "Storage provider not found")
if storage["username"] != username:
if wants_html(request):
return HTMLResponse('Not authorized', status_code=403)
raise HTTPException(403, "Not authorized")
config = storage["config"] if isinstance(storage["config"], dict) else json.loads(storage["config"])
provider = storage_providers.create_provider(storage["provider_type"], {
**config,
"capacity_gb": storage["capacity_gb"]
})
if not provider:
if wants_html(request):
return HTMLResponse('Failed to create provider')
raise HTTPException(500, "Failed to create provider")
success, message = await provider.test_connection()
if wants_html(request):
if success:
return HTMLResponse(f'{message}')
return HTMLResponse(f'{message}')
return {"success": success, "message": message}
STORAGE_PROVIDERS_INFO = {
"pinata": {"name": "Pinata", "desc": "1GB free, IPFS pinning", "color": "blue"},
"web3storage": {"name": "web3.storage", "desc": "IPFS + Filecoin", "color": "green"},
"nftstorage": {"name": "NFT.Storage", "desc": "Free for NFTs", "color": "pink"},
"infura": {"name": "Infura IPFS", "desc": "5GB free", "color": "orange"},
"filebase": {"name": "Filebase", "desc": "5GB free, S3+IPFS", "color": "cyan"},
"storj": {"name": "Storj", "desc": "25GB free", "color": "indigo"},
"local": {"name": "Local Storage", "desc": "Your own disk", "color": "purple"},
}
async def ui_storage_page(username: str, storages: list, request: Request) -> HTMLResponse:
"""Render main storage settings page showing provider types."""
# Count configs per type
type_counts = {}
for s in storages:
pt = s["provider_type"]
type_counts[pt] = type_counts.get(pt, 0) + 1
# Build provider type cards
cards = ""
for ptype, info in STORAGE_PROVIDERS_INFO.items():
count = type_counts.get(ptype, 0)
count_badge = f'{count}' if count > 0 else ""
cards += f'''
'''
# Total stats
total_capacity = sum(s["capacity_gb"] for s in storages)
total_used = sum(s.get("used_bytes", 0) for s in storages)
total_pins = sum(s.get("pin_count", 0) for s in storages)
content = f'''
Storage Providers
Attach your own storage to help power the network. 50% of your capacity is donated to store
shared content, making popular assets more resilient.
{len(storages)}
Total Configs
{total_capacity} GB
Total Capacity
{total_used / (1024**3):.1f} GB
Used
{total_pins}
Total Pins
Select Provider Type
{cards}
'''
return HTMLResponse(base_html("Storage", content, username))
@app.get("/storage/type/{provider_type}")
async def storage_type_page(provider_type: str, request: Request, user: User = Depends(get_optional_user)):
"""Page for managing storage configs of a specific type."""
username = user.username if user else get_user_from_cookie(request)
if not username:
return RedirectResponse(url="/login", status_code=302)
if provider_type not in STORAGE_PROVIDERS_INFO:
raise HTTPException(404, "Invalid provider type")
storages = await db.get_user_storage_by_type(username, provider_type)
# Add usage stats
for storage in storages:
usage = await db.get_storage_usage(storage["id"])
storage["used_bytes"] = usage["used_bytes"]
storage["pin_count"] = usage["pin_count"]
# Mask sensitive config keys
if storage.get("config"):
config = storage["config"] if isinstance(storage["config"], dict) else json.loads(storage["config"])
masked = {}
for k, v in config.items():
if "key" in k.lower() or "token" in k.lower() or "secret" in k.lower():
masked[k] = v[:4] + "..." + v[-4:] if len(str(v)) > 8 else "****"
else:
masked[k] = v
storage["config_display"] = masked
info = STORAGE_PROVIDERS_INFO[provider_type]
return await ui_storage_type_page(username, provider_type, info, storages, request)
async def ui_storage_type_page(username: str, provider_type: str, info: dict, storages: list, request: Request) -> HTMLResponse:
"""Render per-type storage management page."""
def format_bytes(b):
if b > 1024**3:
return f"{b / 1024**3:.1f} GB"
if b > 1024**2:
return f"{b / 1024**2:.1f} MB"
if b > 1024:
return f"{b / 1024:.1f} KB"
return f"{b} bytes"
# Build storage rows
storage_rows = ""
for s in storages:
status_class = "bg-green-600" if s["is_active"] else "bg-gray-600"
status_text = "Active" if s["is_active"] else "Inactive"
config_display = s.get("config_display", {})
config_html = ", ".join(f"{k}: {v}" for k, v in config_display.items() if k != "path")
desc = s.get("description") or ""
desc_html = f'
{desc}
' if desc else ""
storage_rows += f'''
{s["provider_name"] or provider_type}
{desc_html}
{status_text}
Capacity
{s["capacity_gb"]} GB
Donated
{s["capacity_gb"] // 2} GB
Used
{format_bytes(s["used_bytes"])}
Pins
{s["pin_count"]}
{config_html}
'''
if not storages:
storage_rows = f'
No {info["name"]} configs yet. Add one below.
'
# Build form fields based on provider type
form_fields = ""
if provider_type == "pinata":
form_fields = '''
'''
elif provider_type in ("web3storage", "nftstorage"):
form_fields = '''
', content)
# Paragraphs (lines not starting with < or whitespace)
lines = content.split('\n')
result = []
in_paragraph = False
for line in lines:
stripped = line.strip()
if not stripped:
if in_paragraph:
result.append('')
in_paragraph = False
result.append('')
elif stripped.startswith('<'):
if in_paragraph:
result.append('')
in_paragraph = False
result.append(line)
else:
if not in_paragraph:
result.append('
')
in_paragraph = True
result.append(line)
if in_paragraph:
result.append('
')
content = '\n'.join(result)
return content
@app.get("/docs", response_class=HTMLResponse)
async def docs_index(request: Request):
"""Documentation index page."""
user = await get_optional_user(request)
html = f"""
Documentation - Art DAG L2
"""
return HTMLResponse(html)
@app.get("/docs/{doc_name}", response_class=HTMLResponse)
async def docs_page(doc_name: str, request: Request):
"""Render a markdown documentation file as HTML."""
if doc_name not in L2_DOCS_MAP:
raise HTTPException(404, f"Documentation '{doc_name}' not found")
doc_path = L2_DOCS_MAP[doc_name]
if not doc_path.exists():
raise HTTPException(404, f"Documentation file not found: {doc_path}")
content = doc_path.read_text()
html_content = render_markdown(content)
html = f"""
{doc_name.upper()} - Art DAG Documentation