feat: REST API v1 — HMAC-signed product/content creation and file upload
Adds a public/private key pair authentication system and REST API endpoints
for programmatic product and content management. Designed for CI/CD pipelines
(permacomputer.com image hosting).
Auth: HMAC-SHA256 signed requests using public/private key pairs.
The secret key never travels over the wire. Replay window: 300 seconds.
Endpoints:
POST /api/v1/products create product (fiat/crypto priced)
POST /api/v1/content create content (free)
GET /api/v1/products/{id} get product
GET /api/v1/content/{id} get content
POST /api/v1/products/{id}/upload-url presigned S3 POST for direct upload
POST /api/v1/content/{id}/upload-url presigned S3 POST for direct upload
POST /api/v1/products/{id}/files/confirm confirm upload, register metadata
POST /api/v1/content/{id}/files/confirm confirm upload, register metadata
Key management UI in shop settings. Secret shown once on generation.
Migration: mps_api_key table (id, shop_id, public_key, secret_key, label,
created_timestamp, last_used_timestamp, is_active)
Tests: 12 MpsApiKey unit tests, 8 REST API functional tests (269 total passing)
This commit is contained in:
parent
dc06551967
commit
b72fc6bf3b
13 changed files with 1137 additions and 0 deletions
154
docs/tickets/mps-17.md
Normal file
154
docs/tickets/mps-17.md
Normal file
|
|
@ -0,0 +1,154 @@
|
|||
# MPS-17: REST API v1 — HMAC-signed product/content creation + file upload
|
||||
|
||||
## Purpose
|
||||
|
||||
Enable CI/CD pipelines (e.g. permacomputer.com) to programmatically:
|
||||
- Create products (fiat/crypto priced) or content (free)
|
||||
- Upload files directly to Spaces via presigned POST
|
||||
- Confirm uploads and get CDN URLs
|
||||
|
||||
## Auth — HMAC public/private key pairs
|
||||
|
||||
Each shop has API key pairs. A key pair is:
|
||||
- `public_key` — `mps_pub_{32 hex}` — identifies the pair, safe to log
|
||||
- `secret_key` — `mps_sec_{64 hex}` — signs requests, shown **once** on creation
|
||||
|
||||
No bearer tokens. The secret never travels over the wire. Every request is
|
||||
signed with HMAC-SHA256. Replay window: ±300 seconds.
|
||||
|
||||
### Signing scheme
|
||||
|
||||
```
|
||||
string_to_sign = "{METHOD}\n{PATH}\n{TIMESTAMP}\n{SHA256_OF_BODY_HEX}"
|
||||
signature = hmac_sha256(secret_key, string_to_sign).hexdigest()
|
||||
|
||||
Request headers:
|
||||
X-MPS-Key: mps_pub_abc123...
|
||||
X-MPS-Timestamp: 1712345678
|
||||
X-MPS-Signature: sha256=abcdef...
|
||||
```
|
||||
|
||||
### Shell example (for CI)
|
||||
|
||||
```bash
|
||||
METHOD=POST
|
||||
PATH=/api/v1/products
|
||||
TIMESTAMP=$(date +%s)
|
||||
BODY='{"title":"Debian permacomputer","description":"...","price":"0.00"}'
|
||||
BODY_HASH=$(echo -n "$BODY" | sha256sum | awk '{print $1}')
|
||||
STRING_TO_SIGN="${METHOD}\n${PATH}\n${TIMESTAMP}\n${BODY_HASH}"
|
||||
SIG=$(echo -n "$STRING_TO_SIGN" | openssl dgst -sha256 -hmac "$MPS_SECRET_KEY" | awk '{print $2}')
|
||||
|
||||
curl -X POST https://my.makepostsell.com/api/v1/products \
|
||||
-H "X-MPS-Key: $MPS_PUBLIC_KEY" \
|
||||
-H "X-MPS-Timestamp: $TIMESTAMP" \
|
||||
-H "X-MPS-Signature: sha256=$SIG" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d "$BODY"
|
||||
```
|
||||
|
||||
## API Endpoints
|
||||
|
||||
```
|
||||
POST /api/v1/products create product (is_sellable=True)
|
||||
POST /api/v1/content create content (is_sellable=False)
|
||||
GET /api/v1/products/{id} get product
|
||||
GET /api/v1/content/{id} get content
|
||||
POST /api/v1/products/{id}/upload-url get presigned POST URL for file upload
|
||||
POST /api/v1/content/{id}/upload-url get presigned POST URL for file upload
|
||||
POST /api/v1/products/{id}/files/confirm confirm S3 upload, register file, get CDN URL
|
||||
POST /api/v1/content/{id}/files/confirm confirm S3 upload, register file, get CDN URL
|
||||
```
|
||||
|
||||
### POST /api/v1/products
|
||||
|
||||
Request:
|
||||
```json
|
||||
{
|
||||
"title": "Debian permacomputer 6.12 amd64",
|
||||
"description": "Debian bookworm with CWE-407 patched Linux 6.12 kernel",
|
||||
"price": "9.99",
|
||||
"visibility": "public"
|
||||
}
|
||||
```
|
||||
|
||||
Response 201:
|
||||
```json
|
||||
{
|
||||
"id": "abc123...",
|
||||
"url": "https://my.makepostsell.com/p/abc123/debian-permacomputer",
|
||||
"edit_url": "https://my.makepostsell.com/p/abc123/edit"
|
||||
}
|
||||
```
|
||||
|
||||
### POST /api/v1/content
|
||||
|
||||
Same body minus `price`. Response 201 same shape with `/c/` URL.
|
||||
|
||||
### POST /api/v1/products/{id}/upload-url
|
||||
|
||||
Request:
|
||||
```json
|
||||
{
|
||||
"filename": "debian-permacomputer-6.12-amd64.qcow2",
|
||||
"content_type": "application/octet-stream"
|
||||
}
|
||||
```
|
||||
|
||||
Response 200:
|
||||
```json
|
||||
{
|
||||
"upload_url": "https://nyc3.digitaloceanspaces.com/...",
|
||||
"fields": { "key": "...", "AWSAccessKeyId": "...", ... },
|
||||
"confirm_path": "/api/v1/products/{id}/files/confirm",
|
||||
"key": "products/shop_id/product_id/product.qcow2"
|
||||
}
|
||||
```
|
||||
|
||||
### POST /api/v1/products/{id}/files/confirm
|
||||
|
||||
Request:
|
||||
```json
|
||||
{
|
||||
"key": "products/shop_id/product_id/product.qcow2",
|
||||
"filename": "debian-permacomputer-6.12-amd64.qcow2"
|
||||
}
|
||||
```
|
||||
|
||||
Response 200:
|
||||
```json
|
||||
{
|
||||
"cdn_url": "https://plan-period-files.nyc3.cdn.digitaloceanspaces.com/..."
|
||||
}
|
||||
```
|
||||
|
||||
## Shop Settings UI
|
||||
|
||||
New section at bottom of `/s/{shop_id}/settings`:
|
||||
- List active key pairs (label, public key, created date, last used)
|
||||
- "Generate new key pair" form (label input)
|
||||
- Secret shown **once** in a flash-style `<output>` element after generation
|
||||
- Per-key revoke button
|
||||
|
||||
Routes:
|
||||
```
|
||||
POST /s/{shop_id}/api-keys/generate
|
||||
POST /s/{shop_id}/api-keys/{key_id}/revoke
|
||||
```
|
||||
|
||||
## Files
|
||||
|
||||
| File | Change |
|
||||
|------|--------|
|
||||
| `models/api_key.py` | New: `MpsApiKey` model |
|
||||
| `models/__init__.py` | Import `MpsApiKey` |
|
||||
| `models/meta.py` | Add `MpsApiKey` to `CLASS_TO_TABLE` |
|
||||
| `views/api/__init__.py` | HMAC auth: `api_key_required` decorator |
|
||||
| `views/api/items.py` | All API endpoints |
|
||||
| `routes.py` | Add API + key management routes |
|
||||
| `views/shop.py` | `api_keys_generate`, `api_key_revoke` handlers |
|
||||
| `templates/shop_settings.j2` | API Keys section |
|
||||
| `alembic/versions/` | Migration: `mps_api_key` table |
|
||||
| `tests/test_models.py` | `MpsApiKey` unit tests |
|
||||
| `tests/test_integration.py` | HMAC signing integration tests |
|
||||
| `tests/test_functional.py` | API endpoint functional tests |
|
||||
|
|
@ -36,6 +36,8 @@ from .gift_card import *
|
|||
from .gift_card_transaction import *
|
||||
from .cart_gift_card import *
|
||||
|
||||
from .api_key import *
|
||||
|
||||
# run configure_mappers after defining all of the models
|
||||
# to ensure all relationships can be setup.
|
||||
configure_mappers()
|
||||
|
|
|
|||
107
make_post_sell/models/api_key.py
Normal file
107
make_post_sell/models/api_key.py
Normal file
|
|
@ -0,0 +1,107 @@
|
|||
import hmac
|
||||
import hashlib
|
||||
import os
|
||||
import time
|
||||
import uuid
|
||||
|
||||
from sqlalchemy import Column, BigInteger, Boolean, Unicode
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from .meta import (
|
||||
Base,
|
||||
RBase,
|
||||
UUIDType,
|
||||
now_timestamp,
|
||||
foreign_key,
|
||||
)
|
||||
|
||||
|
||||
def _generate_public_key():
|
||||
return "mps_pub_" + os.urandom(16).hex()
|
||||
|
||||
|
||||
def _generate_secret_key():
|
||||
return "mps_sec_" + os.urandom(32).hex()
|
||||
|
||||
|
||||
class MpsApiKey(RBase, Base):
|
||||
"""HMAC key pair for REST API access scoped to a shop.
|
||||
|
||||
The public_key identifies the pair and is safe to log.
|
||||
The secret_key is used to sign requests and is shown once on creation —
|
||||
it is stored plaintext because HMAC verification requires the original value.
|
||||
"""
|
||||
|
||||
__tablename__ = "mps_api_key"
|
||||
|
||||
id = Column(UUIDType, primary_key=True, index=True)
|
||||
shop_id = Column(UUIDType, foreign_key("Shop", "id"), nullable=False, index=True)
|
||||
public_key = Column(Unicode(72), unique=True, nullable=False, index=True)
|
||||
secret_key = Column(Unicode(136), nullable=False)
|
||||
label = Column(Unicode(128), nullable=True)
|
||||
created_timestamp = Column(BigInteger, nullable=False, default=now_timestamp)
|
||||
last_used_timestamp = Column(BigInteger, nullable=True)
|
||||
is_active = Column(Boolean, nullable=False, server_default="1")
|
||||
|
||||
shop = relationship("Shop", back_populates="api_keys")
|
||||
|
||||
@classmethod
|
||||
def generate(cls, shop, label=None):
|
||||
"""Create a new key pair for a shop. Returns (api_key, secret_key_plaintext).
|
||||
|
||||
The secret_key_plaintext is the raw secret — caller must display it
|
||||
once and never retrieve it from the DB again.
|
||||
"""
|
||||
public_key = _generate_public_key()
|
||||
secret_key = _generate_secret_key()
|
||||
key = cls()
|
||||
key.id = uuid.uuid1()
|
||||
key.shop = shop
|
||||
key.public_key = public_key
|
||||
key.secret_key = secret_key
|
||||
key.label = label or ""
|
||||
key.created_timestamp = now_timestamp()
|
||||
key.is_active = True
|
||||
return key, secret_key
|
||||
|
||||
def verify_signature(self, method, path, timestamp_str, body_bytes, signature):
|
||||
"""Verify an HMAC-SHA256 request signature.
|
||||
|
||||
string_to_sign = "{METHOD}\\n{PATH}\\n{TIMESTAMP}\\n{SHA256_OF_BODY_HEX}"
|
||||
signature = "sha256=" + hmac_sha256(secret_key, string_to_sign).hexdigest()
|
||||
|
||||
Returns True if valid, False otherwise (constant-time comparison).
|
||||
"""
|
||||
try:
|
||||
ts = int(timestamp_str)
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
|
||||
# Replay window: ±300 seconds
|
||||
if abs(time.time() - ts) > 300:
|
||||
return False
|
||||
|
||||
body_hash = hashlib.sha256(body_bytes).hexdigest()
|
||||
string_to_sign = f"{method}\n{path}\n{timestamp_str}\n{body_hash}"
|
||||
expected = (
|
||||
"sha256="
|
||||
+ hmac.new(
|
||||
self.secret_key.encode("utf-8"),
|
||||
string_to_sign.encode("utf-8"),
|
||||
hashlib.sha256,
|
||||
).hexdigest()
|
||||
)
|
||||
return hmac.compare_digest(expected, signature)
|
||||
|
||||
@property
|
||||
def masked_secret(self):
|
||||
"""Last 4 chars of secret for display. Never reveal the full value."""
|
||||
return "mps_sec_..." + self.secret_key[-4:]
|
||||
|
||||
|
||||
def get_api_key_by_public_key(dbsession, public_key):
|
||||
return (
|
||||
dbsession.query(MpsApiKey)
|
||||
.filter_by(public_key=public_key, is_active=True)
|
||||
.first()
|
||||
)
|
||||
|
|
@ -46,6 +46,7 @@ CLASS_TO_TABLE = {
|
|||
"GiftCard": "mps_gift_card",
|
||||
"GiftCardTransaction": "mps_gift_card_transaction",
|
||||
"CartGiftCard": "mps_cart_gift_card",
|
||||
"MpsApiKey": "mps_api_key",
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -236,6 +236,10 @@ class Shop(RBase, Base):
|
|||
argument="ShopSearchRequest", lazy="dynamic", back_populates="shop"
|
||||
)
|
||||
|
||||
api_keys = relationship(
|
||||
argument="MpsApiKey", lazy="dynamic", back_populates="shop"
|
||||
)
|
||||
|
||||
def __init__(self, name, phone_number, billing_address, description):
|
||||
self.id = uuid.uuid1()
|
||||
self.name = name
|
||||
|
|
|
|||
|
|
@ -213,3 +213,17 @@ def includeme(config):
|
|||
config.add_route("crypto_cancel", "/crypto/cancel/{payment_id}")
|
||||
config.add_route("crypto_quotes_history", "/u/crypto-quotes")
|
||||
config.add_route("crypto_debug_wallet_scan", "/crypto/debug/wallet-scan")
|
||||
|
||||
# REST API v1 — HMAC-signed
|
||||
config.add_route("api_v1_products", "/api/v1/products")
|
||||
config.add_route("api_v1_product", "/api/v1/products/{product_id}")
|
||||
config.add_route("api_v1_product_upload_url", "/api/v1/products/{product_id}/upload-url")
|
||||
config.add_route("api_v1_product_confirm", "/api/v1/products/{product_id}/files/confirm")
|
||||
config.add_route("api_v1_content", "/api/v1/content")
|
||||
config.add_route("api_v1_content_item", "/api/v1/content/{content_id}")
|
||||
config.add_route("api_v1_content_upload_url", "/api/v1/content/{content_id}/upload-url")
|
||||
config.add_route("api_v1_content_confirm", "/api/v1/content/{content_id}/files/confirm")
|
||||
|
||||
# Shop API key management
|
||||
config.add_route("shop_api_keys_generate", "/s/{shop_id}/api-keys/generate")
|
||||
config.add_route("shop_api_key_revoke", "/s/{shop_id}/api-keys/{key_id}/revoke")
|
||||
|
|
|
|||
|
|
@ -0,0 +1,52 @@
|
|||
"""add mps_api_key table for REST API HMAC auth
|
||||
|
||||
Revision ID: 4f4d7c147437
|
||||
Revises: c0236e351476
|
||||
Create Date: 2026-04-06 15:15:04.347018
|
||||
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
import sqlalchemy_utils.types.uuid
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = '4f4d7c147437'
|
||||
down_revision = 'c0236e351476'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def _table_exists(name):
|
||||
conn = op.get_bind()
|
||||
result = conn.execute(
|
||||
sa.text("SELECT name FROM sqlite_master WHERE type='table' AND name=:name"),
|
||||
{"name": name},
|
||||
)
|
||||
return result.fetchone() is not None
|
||||
|
||||
|
||||
def upgrade():
|
||||
if not _table_exists('mps_api_key'):
|
||||
op.create_table(
|
||||
'mps_api_key',
|
||||
sa.Column('id', sqlalchemy_utils.types.uuid.UUIDType(binary=False), nullable=False),
|
||||
sa.Column('shop_id', sqlalchemy_utils.types.uuid.UUIDType(binary=False), nullable=False),
|
||||
sa.Column('public_key', sa.Unicode(length=72), nullable=False),
|
||||
sa.Column('secret_key', sa.Unicode(length=136), nullable=False),
|
||||
sa.Column('label', sa.Unicode(length=128), nullable=True),
|
||||
sa.Column('created_timestamp', sa.BigInteger(), nullable=False),
|
||||
sa.Column('last_used_timestamp', sa.BigInteger(), nullable=True),
|
||||
sa.Column('is_active', sa.Boolean(), server_default='1', nullable=False),
|
||||
sa.ForeignKeyConstraint(['shop_id'], ['mps_shop.id'], ),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
)
|
||||
op.create_index('ix_mps_api_key_id', 'mps_api_key', ['id'], unique=False)
|
||||
op.create_index('ix_mps_api_key_public_key', 'mps_api_key', ['public_key'], unique=True)
|
||||
op.create_index('ix_mps_api_key_shop_id', 'mps_api_key', ['shop_id'], unique=False)
|
||||
|
||||
|
||||
def downgrade():
|
||||
op.drop_index('ix_mps_api_key_shop_id', table_name='mps_api_key')
|
||||
op.drop_index('ix_mps_api_key_public_key', table_name='mps_api_key')
|
||||
op.drop_index('ix_mps_api_key_id', table_name='mps_api_key')
|
||||
op.drop_table('mps_api_key')
|
||||
|
|
@ -1348,3 +1348,93 @@ function toggleCryptoWallets() {
|
|||
<script src="/static/js/shop-settings.js"></script>
|
||||
|
||||
{%- endblock -%}
|
||||
|
||||
{% block after_content %}
|
||||
<section class="one-column">
|
||||
<section class="shop-settings well">
|
||||
|
||||
<h3>REST API Keys</h3>
|
||||
<p>HMAC-signed key pairs for CI/CD pipelines and server-to-server integrations.
|
||||
Each request is signed with your secret key — it never travels over the wire.</p>
|
||||
|
||||
{% for flash_msg, flash_level in request.session.pop_flash() if flash_level == 'api-secret' %}
|
||||
<div class="well" style="background: var(--color-success-subtle, #f0fdf4); border: 1px solid var(--color-success, #16a34a); padding: var(--space-4);">
|
||||
<strong>Copy your secret key now — it will not be shown again.</strong>
|
||||
<pre style="word-break: break-all; white-space: pre-wrap; margin-top: var(--space-2);">{{ flash_msg }}</pre>
|
||||
</div>
|
||||
{% endfor %}
|
||||
|
||||
{% if api_keys %}
|
||||
<table style="width: 100%; border-collapse: collapse; margin-bottom: var(--space-4);">
|
||||
<thead>
|
||||
<tr>
|
||||
<th style="text-align: left; padding: var(--space-2);">Label</th>
|
||||
<th style="text-align: left; padding: var(--space-2);">Public Key</th>
|
||||
<th style="text-align: left; padding: var(--space-2);">Secret</th>
|
||||
<th style="text-align: left; padding: var(--space-2);">Last Used</th>
|
||||
<th></th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{% for key in api_keys %}
|
||||
<tr>
|
||||
<td style="padding: var(--space-2);">{{ key.label or '—' }}</td>
|
||||
<td style="padding: var(--space-2); font-family: monospace; font-size: 0.85em;">{{ key.public_key }}</td>
|
||||
<td style="padding: var(--space-2); font-family: monospace; font-size: 0.85em; color: var(--color-text-muted);">{{ key.masked_secret }}</td>
|
||||
<td style="padding: var(--space-2); font-size: 0.85em;">
|
||||
{% if key.last_used_timestamp %}
|
||||
{{ key.last_used_timestamp | ago }}
|
||||
{% else %}
|
||||
never
|
||||
{% endif %}
|
||||
</td>
|
||||
<td style="padding: var(--space-2);">
|
||||
<form method="post" action="/s/{{ request.shop.id }}/api-keys/{{ key.id }}/revoke"
|
||||
onsubmit="return confirm('Revoke this key? Any CI using it will stop working.');">
|
||||
<button type="submit" class="mps-btn-danger-sm">Revoke</button>
|
||||
</form>
|
||||
</td>
|
||||
</tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
{% else %}
|
||||
<p style="color: var(--color-text-muted);">No active API keys. Generate one below.</p>
|
||||
{% endif %}
|
||||
|
||||
<form method="post" action="/s/{{ request.shop.id }}/api-keys/generate"
|
||||
onsubmit="submit.disabled = true; return true;">
|
||||
<label for="api_key_label_input">Label (optional)</label>
|
||||
<input
|
||||
name = "label"
|
||||
type = "text"
|
||||
id = "api_key_label_input"
|
||||
class = "full-width-input"
|
||||
placeholder = "e.g. permacomputer CI"
|
||||
maxlength = "128" />
|
||||
<br /><br />
|
||||
<input type="submit" name="submit" class="mps-submit" value="Generate API Key Pair" />
|
||||
</form>
|
||||
|
||||
<br />
|
||||
<details>
|
||||
<summary>How to sign requests</summary>
|
||||
<pre style="font-size: 0.8em; white-space: pre-wrap; word-break: break-all; background: var(--color-surface-alt); padding: var(--space-3); border-radius: var(--radius-md);">METHOD=POST
|
||||
PATH=/api/v1/products
|
||||
TIMESTAMP=$(date +%s)
|
||||
BODY='{"title":"My Item","description":"...","price":"9.99"}'
|
||||
BODY_HASH=$(printf '%s' "$BODY" | sha256sum | awk '{print $1}')
|
||||
STR="${METHOD}\n${PATH}\n${TIMESTAMP}\n${BODY_HASH}"
|
||||
SIG=$(printf '%b' "$STR" | openssl dgst -sha256 -hmac "$MPS_SECRET_KEY" | awk '{print $2}')
|
||||
|
||||
curl -X POST https://my.makepostsell.com/api/v1/products \
|
||||
-H "X-MPS-Key: $MPS_PUBLIC_KEY" \
|
||||
-H "X-MPS-Timestamp: $TIMESTAMP" \
|
||||
-H "X-MPS-Signature: sha256=$SIG" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d "$BODY"</pre>
|
||||
</details>
|
||||
|
||||
</section>
|
||||
</section>
|
||||
{% endblock %}
|
||||
|
|
|
|||
|
|
@ -5059,3 +5059,142 @@ class TestTorrentSettings(_AuthenticatedBase):
|
|||
)
|
||||
# generate_torrent_async must have been called for our product
|
||||
self.assertGreaterEqual(mock_lib.call_count, 1)
|
||||
|
||||
|
||||
class TestRestApiV1(_AuthenticatedBase):
|
||||
"""Functional tests for the HMAC-signed REST API v1."""
|
||||
|
||||
def _make_shop_and_key(self):
|
||||
"""Create a shop for user1, generate an API key, return (shop, public_key, secret_key)."""
|
||||
from ..models.api_key import MpsApiKey
|
||||
|
||||
self.log_in_user(self.user1_creds)
|
||||
redirect_res = self.testapp.post("/s/new", self.shop1_params)
|
||||
res = redirect_res.follow() if redirect_res.status_int == 302 else redirect_res
|
||||
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
|
||||
|
||||
api_key, secret = MpsApiKey.generate(shop, label="test CI")
|
||||
self.dbsession.add(api_key)
|
||||
# Save values before commit detaches the object
|
||||
public_key = api_key.public_key
|
||||
transaction.manager.commit()
|
||||
|
||||
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
|
||||
return shop, public_key, secret
|
||||
|
||||
def _sign(self, public_key, secret, method, path, body_bytes=b""):
|
||||
import hashlib
|
||||
import hmac
|
||||
import time
|
||||
|
||||
timestamp = str(int(time.time()))
|
||||
body_hash = hashlib.sha256(body_bytes).hexdigest()
|
||||
string_to_sign = f"{method}\n{path}\n{timestamp}\n{body_hash}"
|
||||
sig = "sha256=" + hmac.new(
|
||||
secret.encode(), string_to_sign.encode(), hashlib.sha256
|
||||
).hexdigest()
|
||||
return {
|
||||
"X-MPS-Key": public_key,
|
||||
"X-MPS-Timestamp": timestamp,
|
||||
"X-MPS-Signature": sig,
|
||||
}
|
||||
|
||||
def test_create_product_returns_201(self):
|
||||
shop, pub, sec = self._make_shop_and_key()
|
||||
body = b'{"title":"Debian permacomputer","description":"Patched kernel","price":"9.99"}'
|
||||
headers = self._sign(pub, sec, "POST", "/api/v1/products", body)
|
||||
headers["Content-Type"] = "application/json"
|
||||
res = self.testapp.post("/api/v1/products", body, headers=headers, status=201)
|
||||
data = res.json
|
||||
self.assertIn("id", data)
|
||||
self.assertIn("/p/", data["url"])
|
||||
self.assertTrue(data["is_sellable"])
|
||||
|
||||
def test_create_content_returns_201(self):
|
||||
shop, pub, sec = self._make_shop_and_key()
|
||||
body = b'{"title":"Release notes","description":"CWE-407 patch changelog"}'
|
||||
headers = self._sign(pub, sec, "POST", "/api/v1/content", body)
|
||||
headers["Content-Type"] = "application/json"
|
||||
res = self.testapp.post("/api/v1/content", body, headers=headers, status=201)
|
||||
data = res.json
|
||||
self.assertIn("id", data)
|
||||
self.assertIn("/c/", data["url"])
|
||||
self.assertFalse(data["is_sellable"])
|
||||
|
||||
def test_create_product_missing_title_returns_400(self):
|
||||
shop, pub, sec = self._make_shop_and_key()
|
||||
body = b'{"description":"no title","price":"1.00"}'
|
||||
headers = self._sign(pub, sec, "POST", "/api/v1/products", body)
|
||||
headers["Content-Type"] = "application/json"
|
||||
res = self.testapp.post("/api/v1/products", body, headers=headers, status=400)
|
||||
self.assertIn("title", res.json["error"])
|
||||
|
||||
def test_create_product_missing_price_returns_400(self):
|
||||
shop, pub, sec = self._make_shop_and_key()
|
||||
body = b'{"title":"T","description":"D"}'
|
||||
headers = self._sign(pub, sec, "POST", "/api/v1/products", body)
|
||||
headers["Content-Type"] = "application/json"
|
||||
res = self.testapp.post("/api/v1/products", body, headers=headers, status=400)
|
||||
self.assertIn("price", res.json["error"])
|
||||
|
||||
def test_missing_auth_headers_returns_401(self):
|
||||
res = self.testapp.post(
|
||||
"/api/v1/products",
|
||||
b'{"title":"T","description":"D","price":"1.00"}',
|
||||
headers={"Content-Type": "application/json"},
|
||||
status=401,
|
||||
)
|
||||
self.assertIn("error", res.json)
|
||||
|
||||
def test_wrong_signature_returns_401(self):
|
||||
shop, pub, sec = self._make_shop_and_key()
|
||||
import time
|
||||
body = b'{"title":"T","description":"D","price":"1.00"}'
|
||||
headers = {
|
||||
"X-MPS-Key": pub,
|
||||
"X-MPS-Timestamp": str(int(time.time())),
|
||||
"X-MPS-Signature": "sha256=badhex",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
res = self.testapp.post("/api/v1/products", body, headers=headers, status=401)
|
||||
self.assertIn("error", res.json)
|
||||
|
||||
def test_get_product_returns_200(self):
|
||||
shop, pub, sec = self._make_shop_and_key()
|
||||
body = b'{"title":"My Image","description":"A qcow2 image","price":"0.00"}'
|
||||
headers = self._sign(pub, sec, "POST", "/api/v1/products", body)
|
||||
headers["Content-Type"] = "application/json"
|
||||
create_res = self.testapp.post("/api/v1/products", body, headers=headers, status=201)
|
||||
product_id = create_res.json["id"]
|
||||
|
||||
path = f"/api/v1/products/{product_id}"
|
||||
headers = self._sign(pub, sec, "GET", path)
|
||||
res = self.testapp.get(path, headers=headers, status=200)
|
||||
self.assertEqual(res.json["id"], product_id)
|
||||
self.assertEqual(res.json["title"], "My Image")
|
||||
|
||||
def test_get_product_wrong_shop_returns_404(self):
|
||||
"""A key from shop1 cannot access a product in shop2."""
|
||||
from ..models.api_key import MpsApiKey
|
||||
|
||||
# Create shop2 with its own key
|
||||
self.log_in_user(self.user2_creds)
|
||||
redirect_res = self.testapp.post("/s/new", self.shop2_params)
|
||||
shop2 = get_shop_by_name(self.dbsession, self.shop2_params["name"])
|
||||
key2, secret2 = MpsApiKey.generate(shop2, label="shop2 key")
|
||||
self.dbsession.add(key2)
|
||||
pub2 = key2.public_key # save before commit detaches
|
||||
transaction.manager.commit()
|
||||
|
||||
# Create a product in shop2 via API
|
||||
body = b'{"title":"Shop2 Product","description":"desc","price":"5.00"}'
|
||||
headers = self._sign(pub2, secret2, "POST", "/api/v1/products", body)
|
||||
headers["Content-Type"] = "application/json"
|
||||
res = self.testapp.post("/api/v1/products", body, headers=headers, status=201)
|
||||
product_id = res.json["id"]
|
||||
|
||||
# Now shop1 key tries to GET shop2's product
|
||||
shop1, pub1, sec1 = self._make_shop_and_key()
|
||||
path = f"/api/v1/products/{product_id}"
|
||||
headers1 = self._sign(pub1, sec1, "GET", path)
|
||||
self.testapp.get(path, headers=headers1, status=404)
|
||||
|
|
|
|||
|
|
@ -3731,3 +3731,112 @@ class TestTorrentLib(unittest.TestCase):
|
|||
)
|
||||
|
||||
self.assertIsNone(product.torrent_file_url)
|
||||
|
||||
|
||||
class TestMpsApiKey(unittest.TestCase):
|
||||
"""Unit tests for MpsApiKey model."""
|
||||
|
||||
def _make_key(self, label=None):
|
||||
from ..models.api_key import MpsApiKey
|
||||
shop = mock.MagicMock()
|
||||
shop.id = "shop-test-id"
|
||||
key, secret = MpsApiKey.generate(shop, label=label)
|
||||
return key, secret
|
||||
|
||||
def test_generate_returns_key_and_secret(self):
|
||||
key, secret = self._make_key()
|
||||
self.assertIsNotNone(key)
|
||||
self.assertIsNotNone(secret)
|
||||
|
||||
def test_public_key_prefix(self):
|
||||
key, _ = self._make_key()
|
||||
self.assertTrue(key.public_key.startswith("mps_pub_"))
|
||||
|
||||
def test_secret_key_prefix(self):
|
||||
key, secret = self._make_key()
|
||||
self.assertTrue(secret.startswith("mps_sec_"))
|
||||
self.assertEqual(key.secret_key, secret)
|
||||
|
||||
def test_label_stored(self):
|
||||
key, _ = self._make_key(label="permacomputer CI")
|
||||
self.assertEqual(key.label, "permacomputer CI")
|
||||
|
||||
def test_is_active_default(self):
|
||||
key, _ = self._make_key()
|
||||
self.assertTrue(key.is_active)
|
||||
|
||||
def test_masked_secret_format(self):
|
||||
key, secret = self._make_key()
|
||||
masked = key.masked_secret
|
||||
self.assertTrue(masked.startswith("mps_sec_..."))
|
||||
self.assertEqual(masked[-4:], secret[-4:])
|
||||
|
||||
def test_masked_secret_does_not_reveal_full_secret(self):
|
||||
key, secret = self._make_key()
|
||||
self.assertNotIn(secret, key.masked_secret)
|
||||
|
||||
def test_verify_signature_valid(self):
|
||||
import hashlib
|
||||
import hmac
|
||||
import time
|
||||
|
||||
key, secret = self._make_key()
|
||||
method = "POST"
|
||||
path = "/api/v1/products"
|
||||
timestamp = str(int(time.time()))
|
||||
body = b'{"title":"test"}'
|
||||
body_hash = hashlib.sha256(body).hexdigest()
|
||||
string_to_sign = f"{method}\n{path}\n{timestamp}\n{body_hash}"
|
||||
sig = "sha256=" + hmac.new(
|
||||
secret.encode(), string_to_sign.encode(), hashlib.sha256
|
||||
).hexdigest()
|
||||
|
||||
self.assertTrue(key.verify_signature(method, path, timestamp, body, sig))
|
||||
|
||||
def test_verify_signature_wrong_secret(self):
|
||||
import hashlib, hmac, time
|
||||
key, secret = self._make_key()
|
||||
method, path = "POST", "/api/v1/products"
|
||||
timestamp = str(int(time.time()))
|
||||
body = b'{"title":"test"}'
|
||||
body_hash = hashlib.sha256(body).hexdigest()
|
||||
string_to_sign = f"{method}\n{path}\n{timestamp}\n{body_hash}"
|
||||
sig = "sha256=" + hmac.new(
|
||||
b"wrong_secret", string_to_sign.encode(), hashlib.sha256
|
||||
).hexdigest()
|
||||
self.assertFalse(key.verify_signature(method, path, timestamp, body, sig))
|
||||
|
||||
def test_verify_signature_replayed(self):
|
||||
import hashlib, hmac, time
|
||||
key, secret = self._make_key()
|
||||
method, path = "POST", "/api/v1/products"
|
||||
# timestamp 10 minutes in the past
|
||||
timestamp = str(int(time.time()) - 601)
|
||||
body = b'{"title":"test"}'
|
||||
body_hash = hashlib.sha256(body).hexdigest()
|
||||
string_to_sign = f"{method}\n{path}\n{timestamp}\n{body_hash}"
|
||||
sig = "sha256=" + hmac.new(
|
||||
secret.encode(), string_to_sign.encode(), hashlib.sha256
|
||||
).hexdigest()
|
||||
self.assertFalse(key.verify_signature(method, path, timestamp, body, sig))
|
||||
|
||||
def test_verify_signature_tampered_body(self):
|
||||
import hashlib, hmac, time
|
||||
key, secret = self._make_key()
|
||||
method, path = "POST", "/api/v1/products"
|
||||
timestamp = str(int(time.time()))
|
||||
original_body = b'{"title":"test"}'
|
||||
body_hash = hashlib.sha256(original_body).hexdigest()
|
||||
string_to_sign = f"{method}\n{path}\n{timestamp}\n{body_hash}"
|
||||
sig = "sha256=" + hmac.new(
|
||||
secret.encode(), string_to_sign.encode(), hashlib.sha256
|
||||
).hexdigest()
|
||||
# tamper the body before verification
|
||||
tampered_body = b'{"title":"evil"}'
|
||||
self.assertFalse(key.verify_signature(method, path, timestamp, tampered_body, sig))
|
||||
|
||||
def test_unique_keys_per_call(self):
|
||||
key1, secret1 = self._make_key()
|
||||
key2, secret2 = self._make_key()
|
||||
self.assertNotEqual(key1.public_key, key2.public_key)
|
||||
self.assertNotEqual(secret1, secret2)
|
||||
|
|
|
|||
63
make_post_sell/views/api/__init__.py
Normal file
63
make_post_sell/views/api/__init__.py
Normal file
|
|
@ -0,0 +1,63 @@
|
|||
"""MPS REST API v1 — HMAC-signed requests.
|
||||
|
||||
Auth scheme
|
||||
-----------
|
||||
Every request must include three headers:
|
||||
|
||||
X-MPS-Key: mps_pub_{32 hex} — identifies the key pair
|
||||
X-MPS-Timestamp: {unix seconds} — replay protection (±300 s window)
|
||||
X-MPS-Signature: sha256={hex} — HMAC-SHA256 over string_to_sign
|
||||
|
||||
string_to_sign = "{METHOD}\\n{PATH}\\n{TIMESTAMP}\\n{SHA256_OF_BODY_HEX}"
|
||||
|
||||
The secret key signs the request client-side and is verified server-side.
|
||||
It never travels over the wire.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
from pyramid.httpexceptions import HTTPUnauthorized
|
||||
|
||||
from ...models.api_key import get_api_key_by_public_key
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def api_key_required(fn):
|
||||
"""Decorator: verify HMAC signature, attach (api_key, shop) to request."""
|
||||
|
||||
def inner(request):
|
||||
public_key = request.headers.get("X-MPS-Key", "")
|
||||
timestamp = request.headers.get("X-MPS-Timestamp", "")
|
||||
signature = request.headers.get("X-MPS-Signature", "")
|
||||
|
||||
if not public_key or not timestamp or not signature:
|
||||
raise HTTPUnauthorized(
|
||||
json_body={"error": "Missing auth headers: X-MPS-Key, X-MPS-Timestamp, X-MPS-Signature"}
|
||||
)
|
||||
|
||||
api_key = get_api_key_by_public_key(request.dbsession, public_key)
|
||||
if api_key is None:
|
||||
raise HTTPUnauthorized(json_body={"error": "Unknown or revoked API key"})
|
||||
|
||||
body_bytes = request.body if request.body else b""
|
||||
|
||||
if not api_key.verify_signature(
|
||||
request.method,
|
||||
request.path,
|
||||
timestamp,
|
||||
body_bytes,
|
||||
signature,
|
||||
):
|
||||
raise HTTPUnauthorized(json_body={"error": "Invalid signature"})
|
||||
|
||||
# Update last_used without waiting for transaction commit
|
||||
from ...models.meta import now_timestamp
|
||||
api_key.last_used_timestamp = now_timestamp()
|
||||
request.dbsession.add(api_key)
|
||||
|
||||
request.api_key = api_key
|
||||
request.api_shop = api_key.shop
|
||||
return fn(request)
|
||||
|
||||
return inner
|
||||
351
make_post_sell/views/api/items.py
Normal file
351
make_post_sell/views/api/items.py
Normal file
|
|
@ -0,0 +1,351 @@
|
|||
"""MPS REST API v1 — product and content CRUD + file upload.
|
||||
|
||||
Products (is_sellable=True) → fiat or crypto priced items
|
||||
Content (is_sellable=False) → free items
|
||||
|
||||
File upload is a two-step flow:
|
||||
1. POST .../upload-url → returns presigned POST URL + fields (direct to S3)
|
||||
2. Client uploads directly to S3 (file never touches MPS server)
|
||||
3. POST .../files/confirm → registers file metadata, returns CDN URL
|
||||
"""
|
||||
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
|
||||
from pyramid.view import view_config
|
||||
|
||||
from . import api_key_required
|
||||
|
||||
from ...models.product import Product
|
||||
from ...models.meta import now_timestamp
|
||||
|
||||
from ...lib.currency import dollars_to_cents
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Max file size for presigned POST: 10 GB
|
||||
MAX_UPLOAD_BYTES = 10 * 1024 * 1024 * 1024
|
||||
|
||||
VISIBILITY_MAP = {
|
||||
"public": 1,
|
||||
"private": 0,
|
||||
"unlisted": 2,
|
||||
}
|
||||
|
||||
|
||||
def _json_error(request, status, message):
|
||||
request.response.status_code = status
|
||||
return {"error": message}
|
||||
|
||||
|
||||
def _product_to_dict(product, request):
|
||||
p_or_c = "p" if product.is_sellable else "c"
|
||||
return {
|
||||
"id": str(product.id),
|
||||
"title": product.title,
|
||||
"is_sellable": product.is_sellable,
|
||||
"visibility": {0: "private", 1: "public", 2: "unlisted"}.get(
|
||||
product.visibility, "public"
|
||||
),
|
||||
"url": product.absolute_url(request),
|
||||
"edit_url": product.absolute_edit_url(request),
|
||||
}
|
||||
|
||||
|
||||
# ── Create product ────────────────────────────────────────────────────────────
|
||||
|
||||
@view_config(route_name="api_v1_products", renderer="json", request_method="POST")
|
||||
@api_key_required
|
||||
def create_product(request):
|
||||
"""Create a product (is_sellable=True). Requires title, description, price."""
|
||||
try:
|
||||
body = request.json_body
|
||||
except Exception:
|
||||
return _json_error(request, 400, "Invalid JSON body")
|
||||
|
||||
title = (body.get("title") or "").strip()
|
||||
description = (body.get("description") or "").strip()
|
||||
price_str = (body.get("price") or "").strip()
|
||||
visibility_str = (body.get("visibility") or "public").strip()
|
||||
|
||||
if not title:
|
||||
return _json_error(request, 400, "title is required")
|
||||
if not description:
|
||||
return _json_error(request, 400, "description is required")
|
||||
if not price_str:
|
||||
return _json_error(request, 400, "price is required for products")
|
||||
|
||||
try:
|
||||
price_float = float(price_str)
|
||||
if price_float < 0:
|
||||
raise ValueError("price must be >= 0")
|
||||
except ValueError as exc:
|
||||
return _json_error(request, 400, f"Invalid price: {exc}")
|
||||
|
||||
visibility = VISIBILITY_MAP.get(visibility_str, 1)
|
||||
|
||||
shop = request.api_shop
|
||||
|
||||
product = Product(title, description)
|
||||
product.shop = shop
|
||||
product.is_sellable = True
|
||||
product.visibility = visibility
|
||||
|
||||
if product.error_message:
|
||||
return _json_error(request, 400, product.error_message)
|
||||
|
||||
if price_float > 0:
|
||||
product_price = product.set_price(price_float)
|
||||
request.dbsession.add(product_price)
|
||||
|
||||
request.dbsession.add(product)
|
||||
request.dbsession.flush()
|
||||
|
||||
request.response.status_code = 201
|
||||
return _product_to_dict(product, request)
|
||||
|
||||
|
||||
# ── Create content ────────────────────────────────────────────────────────────
|
||||
|
||||
@view_config(route_name="api_v1_content", renderer="json", request_method="POST")
|
||||
@api_key_required
|
||||
def create_content(request):
|
||||
"""Create a content item (is_sellable=False). Requires title, description."""
|
||||
try:
|
||||
body = request.json_body
|
||||
except Exception:
|
||||
return _json_error(request, 400, "Invalid JSON body")
|
||||
|
||||
title = (body.get("title") or "").strip()
|
||||
description = (body.get("description") or "").strip()
|
||||
visibility_str = (body.get("visibility") or "public").strip()
|
||||
|
||||
if not title:
|
||||
return _json_error(request, 400, "title is required")
|
||||
if not description:
|
||||
return _json_error(request, 400, "description is required")
|
||||
|
||||
visibility = VISIBILITY_MAP.get(visibility_str, 1)
|
||||
shop = request.api_shop
|
||||
|
||||
product = Product(title, description)
|
||||
product.shop = shop
|
||||
product.is_sellable = False
|
||||
product.visibility = visibility
|
||||
|
||||
if product.error_message:
|
||||
return _json_error(request, 400, product.error_message)
|
||||
|
||||
request.dbsession.add(product)
|
||||
request.dbsession.flush()
|
||||
|
||||
request.response.status_code = 201
|
||||
return _product_to_dict(product, request)
|
||||
|
||||
|
||||
# ── Get product ───────────────────────────────────────────────────────────────
|
||||
|
||||
@view_config(route_name="api_v1_product", renderer="json", request_method="GET")
|
||||
@api_key_required
|
||||
def get_product(request):
|
||||
product_id = request.matchdict["product_id"]
|
||||
product = _get_item(request, product_id, sellable=True)
|
||||
if product is None:
|
||||
return _json_error(request, 404, "Product not found")
|
||||
return _product_to_dict(product, request)
|
||||
|
||||
|
||||
# ── Get content ───────────────────────────────────────────────────────────────
|
||||
|
||||
@view_config(route_name="api_v1_content_item", renderer="json", request_method="GET")
|
||||
@api_key_required
|
||||
def get_content(request):
|
||||
content_id = request.matchdict["content_id"]
|
||||
product = _get_item(request, content_id, sellable=False)
|
||||
if product is None:
|
||||
return _json_error(request, 404, "Content not found")
|
||||
return _product_to_dict(product, request)
|
||||
|
||||
|
||||
# ── Upload URL — product ──────────────────────────────────────────────────────
|
||||
|
||||
@view_config(route_name="api_v1_product_upload_url", renderer="json", request_method="POST")
|
||||
@api_key_required
|
||||
def product_upload_url(request):
|
||||
product_id = request.matchdict["product_id"]
|
||||
product = _get_item(request, product_id, sellable=True)
|
||||
if product is None:
|
||||
return _json_error(request, 404, "Product not found")
|
||||
return _generate_upload_url(request, product)
|
||||
|
||||
|
||||
# ── Upload URL — content ──────────────────────────────────────────────────────
|
||||
|
||||
@view_config(route_name="api_v1_content_upload_url", renderer="json", request_method="POST")
|
||||
@api_key_required
|
||||
def content_upload_url(request):
|
||||
content_id = request.matchdict["content_id"]
|
||||
product = _get_item(request, content_id, sellable=False)
|
||||
if product is None:
|
||||
return _json_error(request, 404, "Content not found")
|
||||
return _generate_upload_url(request, product)
|
||||
|
||||
|
||||
# ── Confirm upload — product ──────────────────────────────────────────────────
|
||||
|
||||
@view_config(route_name="api_v1_product_confirm", renderer="json", request_method="POST")
|
||||
@api_key_required
|
||||
def product_confirm(request):
|
||||
product_id = request.matchdict["product_id"]
|
||||
product = _get_item(request, product_id, sellable=True)
|
||||
if product is None:
|
||||
return _json_error(request, 404, "Product not found")
|
||||
return _confirm_upload(request, product)
|
||||
|
||||
|
||||
# ── Confirm upload — content ──────────────────────────────────────────────────
|
||||
|
||||
@view_config(route_name="api_v1_content_confirm", renderer="json", request_method="POST")
|
||||
@api_key_required
|
||||
def content_confirm(request):
|
||||
content_id = request.matchdict["content_id"]
|
||||
product = _get_item(request, content_id, sellable=False)
|
||||
if product is None:
|
||||
return _json_error(request, 404, "Content not found")
|
||||
return _confirm_upload(request, product)
|
||||
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def _get_item(request, item_id, sellable):
|
||||
"""Look up a product/content belonging to the authenticated shop."""
|
||||
from ...models.product import get_object_by_id as _get
|
||||
try:
|
||||
product = request.dbsession.query(Product).filter_by(id=item_id).first()
|
||||
except Exception:
|
||||
return None
|
||||
if product is None:
|
||||
return None
|
||||
if str(product.shop_id) != str(request.api_shop.id):
|
||||
return None
|
||||
if product.is_sellable != sellable:
|
||||
return None
|
||||
return product
|
||||
|
||||
|
||||
def _generate_upload_url(request, product):
|
||||
"""Generate a presigned POST URL for direct S3 upload."""
|
||||
try:
|
||||
body = request.json_body
|
||||
except Exception:
|
||||
return _json_error(request, 400, "Invalid JSON body")
|
||||
|
||||
filename = (body.get("filename") or "").strip()
|
||||
content_type = (body.get("content_type") or "application/octet-stream").strip()
|
||||
|
||||
if not filename:
|
||||
return _json_error(request, 400, "filename is required")
|
||||
|
||||
# Sanitize filename
|
||||
safe_filename = "".join(
|
||||
c for c in filename if c.isalnum() or c in ".-_"
|
||||
) or "upload"
|
||||
|
||||
# Tmp key: tmp/{shop_id}/{product_id}/{timestamp}/{safe_filename}
|
||||
tmp_key = (
|
||||
f"tmp/{product.shop_id}/{product.id}"
|
||||
f"/{int(time.time())}/{safe_filename}"
|
||||
)
|
||||
|
||||
shop = request.api_shop
|
||||
client = request.shop_uploads_client
|
||||
bucket = request.shop_bucket_name
|
||||
|
||||
try:
|
||||
presigned = client.generate_presigned_post(
|
||||
Bucket=bucket,
|
||||
Key=tmp_key,
|
||||
ExpiresIn=3600,
|
||||
Conditions=[
|
||||
["content-length-range", 1, MAX_UPLOAD_BYTES],
|
||||
{"Content-Type": content_type},
|
||||
],
|
||||
Fields={"Content-Type": content_type},
|
||||
)
|
||||
except Exception as exc:
|
||||
log.warning("generate_presigned_post failed for product %s: %s", product.id, exc)
|
||||
return _json_error(request, 500, "Failed to generate upload URL")
|
||||
|
||||
p_or_c = "products" if product.is_sellable else "content"
|
||||
return {
|
||||
"upload_url": presigned["url"],
|
||||
"fields": presigned["fields"],
|
||||
"key": tmp_key,
|
||||
"confirm_path": f"/api/v1/{p_or_c}/{product.id}/files/confirm",
|
||||
}
|
||||
|
||||
|
||||
def _confirm_upload(request, product):
|
||||
"""Copy uploaded file from tmp key to product S3 path, register metadata."""
|
||||
try:
|
||||
body = request.json_body
|
||||
except Exception:
|
||||
return _json_error(request, 400, "Invalid JSON body")
|
||||
|
||||
tmp_key = (body.get("key") or "").strip()
|
||||
filename = (body.get("filename") or "").strip()
|
||||
|
||||
if not tmp_key:
|
||||
return _json_error(request, 400, "key is required")
|
||||
if not filename:
|
||||
return _json_error(request, 400, "filename is required")
|
||||
|
||||
# Validate tmp_key belongs to this product (not path traversal)
|
||||
expected_prefix = f"tmp/{product.shop_id}/{product.id}/"
|
||||
if not tmp_key.startswith(expected_prefix):
|
||||
return _json_error(request, 400, "key does not belong to this product")
|
||||
|
||||
# Sanitize filename for S3 key
|
||||
safe_filename = "".join(
|
||||
c for c in filename if c.isalnum() or c in ".-_"
|
||||
) or "upload"
|
||||
|
||||
# Final key: {shop_id}/{product_id}/product.{safe_filename}
|
||||
final_key = f"{product.s3_path}/product.{safe_filename}"
|
||||
|
||||
client = request.shop_uploads_client
|
||||
bucket = request.shop_bucket_name
|
||||
|
||||
# Verify tmp object exists
|
||||
try:
|
||||
client.head_object(Bucket=bucket, Key=tmp_key)
|
||||
except Exception:
|
||||
return _json_error(request, 400, "Uploaded file not found — upload may have failed")
|
||||
|
||||
# Copy to final location
|
||||
try:
|
||||
client.copy_object(
|
||||
CopySource={"Bucket": bucket, "Key": tmp_key},
|
||||
Bucket=bucket,
|
||||
Key=final_key,
|
||||
)
|
||||
except Exception as exc:
|
||||
log.warning("copy_object failed for product %s: %s", product.id, exc)
|
||||
return _json_error(request, 500, "Failed to register uploaded file")
|
||||
|
||||
# Register metadata on product
|
||||
product.store_file_metadata(final_key)
|
||||
request.dbsession.add(product)
|
||||
|
||||
# Delete tmp object
|
||||
try:
|
||||
client.delete_object(Bucket=bucket, Key=tmp_key)
|
||||
except Exception as exc:
|
||||
log.warning("delete tmp key failed for product %s: %s", product.id, exc)
|
||||
|
||||
# Build CDN URL
|
||||
cdn_endpoint = request.shop_cdn_endpoint
|
||||
cdn_url = f"{cdn_endpoint.rstrip('/')}/{final_key}"
|
||||
|
||||
return {"cdn_url": cdn_url}
|
||||
|
|
@ -1438,4 +1438,55 @@ def shop_settings(request):
|
|||
"torrent_enabled": shop.torrent_enabled,
|
||||
"signed_posts": signed_posts,
|
||||
"get_endpoints": get_endpoints,
|
||||
"api_keys": shop.api_keys.filter_by(is_active=True).order_by("created_timestamp").all(),
|
||||
}
|
||||
|
||||
|
||||
# ── API key management ────────────────────────────────────────────────────────
|
||||
|
||||
@view_config(route_name="shop_api_keys_generate", renderer="json", request_method="POST")
|
||||
@user_required()
|
||||
@shop_owner_required()
|
||||
def api_keys_generate(request):
|
||||
"""Generate a new HMAC key pair for the shop. Returns secret once."""
|
||||
from ..models.api_key import MpsApiKey
|
||||
|
||||
label = request.params.get("label", "").strip()[:128]
|
||||
shop = request.shop
|
||||
|
||||
api_key, secret_plaintext = MpsApiKey.generate(shop, label=label)
|
||||
request.dbsession.add(api_key)
|
||||
request.dbsession.flush()
|
||||
|
||||
# Flash the secret — shown once, never retrievable again
|
||||
request.session.flash((
|
||||
f"API key generated. Public key: {api_key.public_key} "
|
||||
f"Secret (copy now — shown once): {secret_plaintext}",
|
||||
"api-secret",
|
||||
))
|
||||
return HTTPFound(request.route_url("shop_settings", shop_id=shop.id))
|
||||
|
||||
|
||||
@view_config(route_name="shop_api_key_revoke", renderer="json", request_method="POST")
|
||||
@user_required()
|
||||
@shop_owner_required()
|
||||
def api_key_revoke(request):
|
||||
"""Revoke (deactivate) an API key."""
|
||||
from ..models.api_key import MpsApiKey
|
||||
|
||||
key_id = request.matchdict["key_id"]
|
||||
shop = request.shop
|
||||
|
||||
api_key = (
|
||||
request.dbsession.query(MpsApiKey)
|
||||
.filter_by(id=key_id, shop_id=shop.id, is_active=True)
|
||||
.first()
|
||||
)
|
||||
if api_key is None:
|
||||
request.session.flash(("API key not found.", "error"))
|
||||
return HTTPFound(request.route_url("shop_settings", shop_id=shop.id))
|
||||
|
||||
api_key.is_active = False
|
||||
request.dbsession.add(api_key)
|
||||
request.session.flash(("API key revoked.", "success"))
|
||||
return HTTPFound(request.route_url("shop_settings", shop_id=shop.id))
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue