kqjy/MyFSIO
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 13 Wiki Activity

Fix routing conflicts: move admin endpoints to reserved paths

Browse Source
This commit is contained in:
kqjy
2026-01-18 21:35:39 +08:00
parent 5ab62a00ff
commit ebc315c1cc
8 changed files with 119 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Dockerfile
View File

@@ -32,6 +32,6 @@ ENV APP_HOST=0.0.0.0 \
FLASK_DEBUG=0 FLASK_DEBUG=0
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \ HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
CMD python -c "import requests; requests.get('http://localhost:5000/healthz', timeout=2)" CMD python -c "import requests; requests.get('http://localhost:5000/myfsio/health', timeout=2)"
CMD ["./docker-entrypoint.sh"] CMD ["./docker-entrypoint.sh"]

16
README.md
View File

@@ -149,19 +149,13 @@ All endpoints require AWS Signature Version 4 authentication unless using presig
| `POST` | `/<bucket>/<key>?uploadId=X` | Complete multipart upload | | `POST` | `/<bucket>/<key>?uploadId=X` | Complete multipart upload |
| `DELETE` | `/<bucket>/<key>?uploadId=X` | Abort multipart upload | | `DELETE` | `/<bucket>/<key>?uploadId=X` | Abort multipart upload |
### Presigned URLs ### Bucket Policies (S3-compatible)
| Method | Endpoint | Description | | Method | Endpoint | Description |
|--------|----------|-------------| |--------|----------|-------------|
| `POST` | `/presign/<bucket>/<key>` | Generate presigned URL | | `GET` | `/<bucket>?policy` | Get bucket policy |
| `PUT` | `/<bucket>?policy` | Set bucket policy |
### Bucket Policies | `DELETE` | `/<bucket>?policy` | Delete bucket policy |
| Method | Endpoint | Description |
|--------|----------|-------------|
| `GET` | `/bucket-policy/<bucket>` | Get bucket policy |
| `PUT` | `/bucket-policy/<bucket>` | Set bucket policy |
| `DELETE` | `/bucket-policy/<bucket>` | Delete bucket policy |
### Versioning ### Versioning
@@ -175,7 +169,7 @@ All endpoints require AWS Signature Version 4 authentication unless using presig
| Method | Endpoint | Description | | Method | Endpoint | Description |
|--------|----------|-------------| |--------|----------|-------------|
| `GET` | `/healthz` | Health check endpoint | | `GET` | `/myfsio/health` | Health check endpoint |
## IAM & Access Control ## IAM & Access Control

2
app/__init__.py
View File

@@ -278,7 +278,7 @@ def create_app(
return render_template("404.html"), 404 return render_template("404.html"), 404
return error return error
@app.get("/healthz") @app.get("/myfsio/health")
def healthcheck() -> Dict[str, str]: def healthcheck() -> Dict[str, str]:
return {"status": "ok"} return {"status": "ok"}

55
app/s3_api.py
View File

@@ -590,6 +590,7 @@ def _generate_presigned_url(
bucket_name: str, bucket_name: str,
object_key: str, object_key: str,
expires_in: int, expires_in: int,
api_base_url: str | None = None,
) -> str: ) -> str:
region = current_app.config["AWS_REGION"] region = current_app.config["AWS_REGION"]
service = current_app.config["AWS_SERVICE"] service = current_app.config["AWS_SERVICE"]
@@ -610,7 +611,7 @@ def _generate_presigned_url(
} }
canonical_query = _encode_query_params(query_params) canonical_query = _encode_query_params(query_params)
api_base = current_app.config.get("API_BASE_URL") api_base = api_base_url or current_app.config.get("API_BASE_URL")
if api_base: if api_base:
parsed = urlparse(api_base) parsed = urlparse(api_base)
host = parsed.netloc host = parsed.netloc
@@ -940,6 +941,7 @@ def _maybe_handle_bucket_subresource(bucket_name: str) -> Response | None:
"notification": _bucket_notification_handler, "notification": _bucket_notification_handler,
"logging": _bucket_logging_handler, "logging": _bucket_logging_handler,
"uploads": _bucket_uploads_handler, "uploads": _bucket_uploads_handler,
"policy": _bucket_policy_handler,
} }
requested = [key for key in handlers if key in request.args] requested = [key for key in handlers if key in request.args]
if not requested: if not requested:
@@ -2642,9 +2644,9 @@ def _list_parts(bucket_name: str, object_key: str) -> Response:
return _xml_response(root) return _xml_response(root)
@s3_api_bp.route("/bucket-policy/<bucket_name>", methods=["GET", "PUT", "DELETE"]) def _bucket_policy_handler(bucket_name: str) -> Response:
@limiter.limit("30 per minute") if request.method not in {"GET", "PUT", "DELETE"}:
def bucket_policy_handler(bucket_name: str) -> Response: return _method_not_allowed(["GET", "PUT", "DELETE"])
principal, error = _require_principal() principal, error = _require_principal()
if error: if error:
return error return error
@@ -2676,51 +2678,6 @@ def bucket_policy_handler(bucket_name: str) -> Response:
return Response(status=204) return Response(status=204)
@s3_api_bp.post("/presign/<bucket_name>/<path:object_key>")
@limiter.limit("45 per minute")
def presign_object(bucket_name: str, object_key: str):
payload = request.get_json(silent=True) or {}
method = str(payload.get("method", "GET")).upper()
allowed_methods = {"GET", "PUT", "DELETE"}
if method not in allowed_methods:
return _error_response("InvalidRequest", "Method must be GET, PUT, or DELETE", 400)
try:
expires = int(payload.get("expires_in", 900))
except (TypeError, ValueError):
return _error_response("InvalidRequest", "expires_in must be an integer", 400)
expires = max(1, min(expires, 7 * 24 * 3600))
action = "read" if method == "GET" else ("delete" if method == "DELETE" else "write")
principal, error = _require_principal()
if error:
return error
try:
_authorize_action(principal, bucket_name, action, object_key=object_key)
except IamError as exc:
return _error_response("AccessDenied", str(exc), 403)
storage = _storage()
if not storage.bucket_exists(bucket_name):
return _error_response("NoSuchBucket", "Bucket does not exist", 404)
if action != "write":
try:
storage.get_object_path(bucket_name, object_key)
except StorageError:
return _error_response("NoSuchKey", "Object not found", 404)
secret = _iam().secret_for_key(principal.access_key)
url = _generate_presigned_url(
principal=principal,
secret_key=secret,
method=method,
bucket_name=bucket_name,
object_key=object_key,
expires_in=expires,
)
current_app.logger.info(
"Presigned URL generated",
extra={"bucket": bucket_name, "key": object_key, "method": method},
)
return jsonify({"url": url, "method": method, "expires_in": expires})
@s3_api_bp.route("/<bucket_name>", methods=["HEAD"]) @s3_api_bp.route("/<bucket_name>", methods=["HEAD"])
@limiter.limit("100 per minute") @limiter.limit("100 per minute")
def head_bucket(bucket_name: str) -> Response: def head_bucket(bucket_name: str) -> Response:

62
app/ui.py
View File

@@ -36,6 +36,7 @@ from .extensions import limiter, csrf
from .iam import IamError from .iam import IamError
from .kms import KMSManager from .kms import KMSManager
from .replication import ReplicationManager, ReplicationRule from .replication import ReplicationManager, ReplicationRule
from .s3_api import _generate_presigned_url
from .secret_store import EphemeralSecretStore from .secret_store import EphemeralSecretStore
from .storage import ObjectStorage, StorageError from .storage import ObjectStorage, StorageError
@@ -1135,42 +1136,43 @@ def object_presign(bucket_name: str, object_key: str):
principal = _current_principal() principal = _current_principal()
payload = request.get_json(silent=True) or {} payload = request.get_json(silent=True) or {}
method = str(payload.get("method", "GET")).upper() method = str(payload.get("method", "GET")).upper()
allowed_methods = {"GET", "PUT", "DELETE"}
if method not in allowed_methods:
return jsonify({"error": "Method must be GET, PUT, or DELETE"}), 400
action = "read" if method == "GET" else ("delete" if method == "DELETE" else "write") action = "read" if method == "GET" else ("delete" if method == "DELETE" else "write")
try: try:
_authorize_ui(principal, bucket_name, action, object_key=object_key) _authorize_ui(principal, bucket_name, action, object_key=object_key)
except IamError as exc: except IamError as exc:
return jsonify({"error": str(exc)}), 403 return jsonify({"error": str(exc)}), 403
try:
expires = int(payload.get("expires_in", 900))
except (TypeError, ValueError):
return jsonify({"error": "expires_in must be an integer"}), 400
expires = max(1, min(expires, 7 * 24 * 3600))
storage = _storage()
if not storage.bucket_exists(bucket_name):
return jsonify({"error": "Bucket does not exist"}), 404
if action != "write":
try:
storage.get_object_path(bucket_name, object_key)
except StorageError:
return jsonify({"error": "Object not found"}), 404
secret = _iam().secret_for_key(principal.access_key)
api_base = current_app.config.get("API_BASE_URL") or "http://127.0.0.1:5000" api_base = current_app.config.get("API_BASE_URL") or "http://127.0.0.1:5000"
api_base = api_base.rstrip("/") url = _generate_presigned_url(
encoded_key = quote(object_key, safe="/") principal=principal,
url = f"{api_base}/presign/{bucket_name}/{encoded_key}" secret_key=secret,
method=method,
parsed_api = urlparse(api_base) bucket_name=bucket_name,
headers = _api_headers() object_key=object_key,
headers["X-Forwarded-Host"] = parsed_api.netloc or "127.0.0.1:5000" expires_in=expires,
headers["X-Forwarded-Proto"] = parsed_api.scheme or "http" api_base_url=api_base,
headers["X-Forwarded-For"] = request.remote_addr or "127.0.0.1" )
current_app.logger.info(
try: "Presigned URL generated",
response = requests.post(url, headers=headers, json=payload, timeout=5) extra={"bucket": bucket_name, "key": object_key, "method": method},
except requests.RequestException as exc: )
return jsonify({"error": f"API unavailable: {exc}"}), 502 return jsonify({"url": url, "method": method, "expires_in": expires})
try:
body = response.json()
except ValueError:
text = response.text or ""
if text.strip().startswith("<"):
import xml.etree.ElementTree as ET
try:
root = ET.fromstring(text)
message = root.findtext(".//Message") or root.findtext(".//Code") or "Unknown S3 error"
body = {"error": message}
except ET.ParseError:
body = {"error": text or "API returned an empty response"}
else:
body = {"error": text or "API returned an empty response"}
return jsonify(body), response.status_code
@ui_bp.get("/buckets/<bucket_name>/objects/<path:object_key>/metadata") @ui_bp.get("/buckets/<bucket_name>/objects/<path:object_key>/metadata")

28
docs.md
View File

@@ -122,7 +122,7 @@ With these volumes attached you can rebuild/restart the container without losing
### Versioning ### Versioning
The repo now tracks a human-friendly release string inside `app/version.py` (see the `APP_VERSION` constant). Edit that value whenever you cut a release. The constant flows into Flask as `APP_VERSION` and is exposed via `GET /healthz`, so you can monitor deployments or surface it in UIs. The repo now tracks a human-friendly release string inside `app/version.py` (see the `APP_VERSION` constant). Edit that value whenever you cut a release. The constant flows into Flask as `APP_VERSION` and is exposed via `GET /myfsio/health`, so you can monitor deployments or surface it in UIs.
## 3. Configuration Reference ## 3. Configuration Reference
@@ -277,14 +277,14 @@ The application automatically trusts these headers to generate correct presigned
### Version Checking ### Version Checking
The application version is tracked in `app/version.py` and exposed via: The application version is tracked in `app/version.py` and exposed via:
- **Health endpoint:** `GET /healthz` returns JSON with `version` field - **Health endpoint:** `GET /myfsio/health` returns JSON with `version` field
- **Metrics dashboard:** Navigate to `/ui/metrics` to see the running version in the System Status card - **Metrics dashboard:** Navigate to `/ui/metrics` to see the running version in the System Status card
To check your current version: To check your current version:
```bash ```bash
# API health endpoint # API health endpoint
curl http://localhost:5000/healthz curl http://localhost:5000/myfsio/health
# Or inspect version.py directly # Or inspect version.py directly
cat app/version.py | grep APP_VERSION cat app/version.py | grep APP_VERSION
@@ -377,7 +377,7 @@ docker run -d \
myfsio:latest myfsio:latest
# 5. Verify health # 5. Verify health
curl http://localhost:5000/healthz curl http://localhost:5000/myfsio/health
``` ```
### Version Compatibility Checks ### Version Compatibility Checks
@@ -502,7 +502,7 @@ docker run -d \
myfsio:0.1.3 # specify previous version tag myfsio:0.1.3 # specify previous version tag
# 3. Verify # 3. Verify
curl http://localhost:5000/healthz curl http://localhost:5000/myfsio/health
``` ```
#### Emergency Config Restore #### Emergency Config Restore
@@ -528,7 +528,7 @@ For production environments requiring zero downtime:
APP_PORT=5001 UI_PORT=5101 python run.py & APP_PORT=5001 UI_PORT=5101 python run.py &
# 2. Health check new instance # 2. Health check new instance
curl http://localhost:5001/healthz curl http://localhost:5001/myfsio/health
# 3. Update load balancer to route to new ports # 3. Update load balancer to route to new ports
@@ -544,7 +544,7 @@ After any update, verify functionality:
```bash ```bash
# 1. Health check # 1. Health check
curl http://localhost:5000/healthz curl http://localhost:5000/myfsio/health
# 2. Login to UI # 2. Login to UI
open http://localhost:5100/ui open http://localhost:5100/ui
@@ -588,7 +588,7 @@ APP_PID=$!
# Wait and health check # Wait and health check
sleep 5 sleep 5
if curl -f http://localhost:5000/healthz; then if curl -f http://localhost:5000/myfsio/health; then
echo "Update successful!" echo "Update successful!"
else else
echo "Health check failed, rolling back..." echo "Health check failed, rolling back..."
@@ -860,7 +860,7 @@ A request is allowed only if:
### Editing via CLI ### Editing via CLI
```bash ```bash
curl -X PUT http://127.0.0.1:5000/bucket-policy/test \ curl -X PUT "http://127.0.0.1:5000/test?policy" \
-H "Content-Type: application/json" \ -H "Content-Type: application/json" \
-H "X-Access-Key: ..." -H "X-Secret-Key: ..." \ -H "X-Access-Key: ..." -H "X-Secret-Key: ..." \
-d '{ -d '{
@@ -923,9 +923,8 @@ Drag files directly onto the objects table to upload them to the current bucket
## 6. Presigned URLs ## 6. Presigned URLs
- Trigger from the UI using the **Presign** button after selecting an object. - Trigger from the UI using the **Presign** button after selecting an object.
- Or call `POST /presign/<bucket>/<key>` with JSON `{ "method": "GET", "expires_in": 900 }`.
- Supported methods: `GET`, `PUT`, `DELETE`; expiration must be `1..604800` seconds. - Supported methods: `GET`, `PUT`, `DELETE`; expiration must be `1..604800` seconds.
- The service signs requests using the caller’s IAM credentials and enforces bucket policies both when issuing and when the presigned URL is used. - The service signs requests using the caller's IAM credentials and enforces bucket policies both when issuing and when the presigned URL is used.
- Legacy share links have been removed; presigned URLs now handle both private and public workflows. - Legacy share links have been removed; presigned URLs now handle both private and public workflows.
### Multipart Upload Example ### Multipart Upload Example
@@ -1314,10 +1313,9 @@ GET /<bucket> # List objects
PUT /<bucket>/<key> # Upload object PUT /<bucket>/<key> # Upload object
GET /<bucket>/<key> # Download object GET /<bucket>/<key> # Download object
DELETE /<bucket>/<key> # Delete object DELETE /<bucket>/<key> # Delete object
POST /presign/<bucket>/<key> # Generate SigV4 URL GET /<bucket>?policy # Fetch policy
GET /bucket-policy/<bucket> # Fetch policy PUT /<bucket>?policy # Upsert policy
PUT /bucket-policy/<bucket> # Upsert policy DELETE /<bucket>?policy # Delete policy
DELETE /bucket-policy/<bucket> # Delete policy
GET /<bucket>?quota # Get bucket quota GET /<bucket>?quota # Get bucket quota
PUT /<bucket>?quota # Set bucket quota (admin only) PUT /<bucket>?quota # Set bucket quota (admin only)
``` ```

35
templates/docs.html
View File

@@ -375,11 +375,8 @@ curl -X PUT {{ api_base }}/demo/notes.txt \
-H "X-Secret-Key: &lt;secret_key&gt;" \ -H "X-Secret-Key: &lt;secret_key&gt;" \
--data-binary @notes.txt --data-binary @notes.txt
curl -X POST {{ api_base }}/presign/demo/notes.txt \ # Presigned URLs are generated via the UI
-H "Content-Type: application/json" \ # Use the "Presign" button in the object browser
-H "X-Access-Key: &lt;access_key&gt;" \
-H "X-Secret-Key: &lt;secret_key&gt;" \
-d '{"method":"GET", "expires_in": 900}'
</code></pre> </code></pre>
</div> </div>
</div> </div>
@@ -437,13 +434,8 @@ curl -X POST {{ api_base }}/presign/demo/notes.txt \
</tr> </tr>
<tr> <tr>
<td>GET/PUT/DELETE</td> <td>GET/PUT/DELETE</td>
<td><code>/bucket-policy/&lt;bucket&gt;</code></td> <td><code>/&lt;bucket&gt;?policy</code></td>
<td>Fetch, upsert, or remove a bucket policy.</td> <td>Fetch, upsert, or remove a bucket policy (S3-compatible).</td>
</tr>
<tr>
<td>POST</td>
<td><code>/presign/&lt;bucket&gt;/&lt;key&gt;</code></td>
<td>Generate SigV4 URLs for GET/PUT/DELETE with custom expiry.</td>
</tr> </tr>
</tbody> </tbody>
</table> </table>
@@ -542,17 +534,16 @@ s3.complete_multipart_upload(
)</code></pre> )</code></pre>
<h3 class="h6 text-uppercase text-muted mt-4">Presigned URLs for Sharing</h3> <h3 class="h6 text-uppercase text-muted mt-4">Presigned URLs for Sharing</h3>
<pre class="mb-0"><code class="language-bash"># Generate a download link valid for 15 minutes <pre class="mb-0"><code class="language-text"># Generate presigned URLs via the UI:
curl -X POST "{{ api_base }}/presign/mybucket/photo.jpg" \ # 1. Navigate to your bucket in the object browser
-H "Content-Type: application/json" \ # 2. Select the object you want to share
-H "X-Access-Key: &lt;key&gt;" -H "X-Secret-Key: &lt;secret&gt;" \ # 3. Click the "Presign" button
-d '{"method": "GET", "expires_in": 900}' # 4. Choose method (GET/PUT/DELETE) and expiration time
# 5. Copy the generated URL
# Generate an upload link (PUT) valid for 1 hour # Supported options:
curl -X POST "{{ api_base }}/presign/mybucket/upload.bin" \ # - Method: GET (download), PUT (upload), DELETE (remove)
-H "Content-Type: application/json" \ # - Expiration: 1 second to 7 days (604800 seconds)</code></pre>
-H "X-Access-Key: &lt;key&gt;" -H "X-Secret-Key: &lt;secret&gt;" \
-d '{"method": "PUT", "expires_in": 3600}'</code></pre>
</div> </div>
</article> </article>
<article id="replication" class="card shadow-sm docs-section"> <article id="replication" class="card shadow-sm docs-section">

115
tests/test_api.py
View File

@@ -1,6 +1,3 @@
from urllib.parse import urlsplit
def test_bucket_and_object_lifecycle(client, signer): def test_bucket_and_object_lifecycle(client, signer):
headers = signer("PUT", "/photos") headers = signer("PUT", "/photos")
response = client.put("/photos", headers=headers) response = client.put("/photos", headers=headers)
@@ -105,7 +102,7 @@ def test_request_id_header_present(client, signer):
def test_healthcheck_returns_status(client): def test_healthcheck_returns_status(client):
response = client.get("/healthz") response = client.get("/myfsio/health")
data = response.get_json() data = response.get_json()
assert response.status_code == 200 assert response.status_code == 200
assert data["status"] == "ok" assert data["status"] == "ok"
@@ -117,36 +114,20 @@ def test_missing_credentials_denied(client):
assert response.status_code == 403 assert response.status_code == 403
def test_presign_and_bucket_policies(client, signer): def test_bucket_policies_deny_reads(client, signer):
# Create bucket and object import json
headers = signer("PUT", "/docs") headers = signer("PUT", "/docs")
assert client.put("/docs", headers=headers).status_code == 200 assert client.put("/docs", headers=headers).status_code == 200
headers = signer("PUT", "/docs/readme.txt", body=b"content") headers = signer("PUT", "/docs/readme.txt", body=b"content")
assert client.put("/docs/readme.txt", headers=headers, data=b"content").status_code == 200 assert client.put("/docs/readme.txt", headers=headers, data=b"content").status_code == 200
# Generate presigned GET URL and follow it headers = signer("GET", "/docs/readme.txt")
json_body = {"method": "GET", "expires_in": 120} response = client.get("/docs/readme.txt", headers=headers)
# Flask test client json parameter automatically sets Content-Type and serializes body
# But for signing we need the body bytes.
import json
body_bytes = json.dumps(json_body).encode("utf-8")
headers = signer("POST", "/presign/docs/readme.txt", headers={"Content-Type": "application/json"}, body=body_bytes)
response = client.post(
"/presign/docs/readme.txt",
headers=headers,
json=json_body,
)
assert response.status_code == 200 assert response.status_code == 200
presigned_url = response.get_json()["url"] assert response.data == b"content"
parts = urlsplit(presigned_url)
presigned_path = f"{parts.path}?{parts.query}"
download = client.get(presigned_path)
assert download.status_code == 200
assert download.data == b"content"
# Attach a deny policy for GETs
policy = { policy = {
"Version": "2012-10-17", "Version": "2012-10-17",
"Statement": [ "Statement": [
@@ -160,29 +141,26 @@ def test_presign_and_bucket_policies(client, signer):
], ],
} }
policy_bytes = json.dumps(policy).encode("utf-8") policy_bytes = json.dumps(policy).encode("utf-8")
headers = signer("PUT", "/bucket-policy/docs", headers={"Content-Type": "application/json"}, body=policy_bytes) headers = signer("PUT", "/docs?policy", headers={"Content-Type": "application/json"}, body=policy_bytes)
assert client.put("/bucket-policy/docs", headers=headers, json=policy).status_code == 204 assert client.put("/docs?policy", headers=headers, json=policy).status_code == 204
headers = signer("GET", "/bucket-policy/docs") headers = signer("GET", "/docs?policy")
fetched = client.get("/bucket-policy/docs", headers=headers) fetched = client.get("/docs?policy", headers=headers)
assert fetched.status_code == 200 assert fetched.status_code == 200
assert fetched.get_json()["Version"] == "2012-10-17" assert fetched.get_json()["Version"] == "2012-10-17"
# Reads are now denied by bucket policy
headers = signer("GET", "/docs/readme.txt") headers = signer("GET", "/docs/readme.txt")
denied = client.get("/docs/readme.txt", headers=headers) denied = client.get("/docs/readme.txt", headers=headers)
assert denied.status_code == 403 assert denied.status_code == 403
# Presign attempts are also denied headers = signer("DELETE", "/docs?policy")
json_body = {"method": "GET", "expires_in": 60} assert client.delete("/docs?policy", headers=headers).status_code == 204
body_bytes = json.dumps(json_body).encode("utf-8")
headers = signer("POST", "/presign/docs/readme.txt", headers={"Content-Type": "application/json"}, body=body_bytes) headers = signer("DELETE", "/docs/readme.txt")
response = client.post( assert client.delete("/docs/readme.txt", headers=headers).status_code == 204
"/presign/docs/readme.txt",
headers=headers, headers = signer("DELETE", "/docs")
json=json_body, assert client.delete("/docs", headers=headers).status_code == 204
)
assert response.status_code == 403
def test_trailing_slash_returns_xml(client): def test_trailing_slash_returns_xml(client):
@@ -193,9 +171,11 @@ def test_trailing_slash_returns_xml(client):
def test_public_policy_allows_anonymous_list_and_read(client, signer): def test_public_policy_allows_anonymous_list_and_read(client, signer):
import json
headers = signer("PUT", "/public") headers = signer("PUT", "/public")
assert client.put("/public", headers=headers).status_code == 200 assert client.put("/public", headers=headers).status_code == 200
headers = signer("PUT", "/public/hello.txt", body=b"hi") headers = signer("PUT", "/public/hello.txt", body=b"hi")
assert client.put("/public/hello.txt", headers=headers, data=b"hi").status_code == 200 assert client.put("/public/hello.txt", headers=headers, data=b"hi").status_code == 200
@@ -221,10 +201,9 @@ def test_public_policy_allows_anonymous_list_and_read(client, signer):
}, },
], ],
} }
import json
policy_bytes = json.dumps(policy).encode("utf-8") policy_bytes = json.dumps(policy).encode("utf-8")
headers = signer("PUT", "/bucket-policy/public", headers={"Content-Type": "application/json"}, body=policy_bytes) headers = signer("PUT", "/public?policy", headers={"Content-Type": "application/json"}, body=policy_bytes)
assert client.put("/bucket-policy/public", headers=headers, json=policy).status_code == 204 assert client.put("/public?policy", headers=headers, json=policy).status_code == 204
list_response = client.get("/public") list_response = client.get("/public")
assert list_response.status_code == 200 assert list_response.status_code == 200
@@ -236,18 +215,20 @@ def test_public_policy_allows_anonymous_list_and_read(client, signer):
headers = signer("DELETE", "/public/hello.txt") headers = signer("DELETE", "/public/hello.txt")
assert client.delete("/public/hello.txt", headers=headers).status_code == 204 assert client.delete("/public/hello.txt", headers=headers).status_code == 204
headers = signer("DELETE", "/bucket-policy/public") headers = signer("DELETE", "/public?policy")
assert client.delete("/bucket-policy/public", headers=headers).status_code == 204 assert client.delete("/public?policy", headers=headers).status_code == 204
headers = signer("DELETE", "/public") headers = signer("DELETE", "/public")
assert client.delete("/public", headers=headers).status_code == 204 assert client.delete("/public", headers=headers).status_code == 204
def test_principal_dict_with_object_get_only(client, signer): def test_principal_dict_with_object_get_only(client, signer):
import json
headers = signer("PUT", "/mixed") headers = signer("PUT", "/mixed")
assert client.put("/mixed", headers=headers).status_code == 200 assert client.put("/mixed", headers=headers).status_code == 200
headers = signer("PUT", "/mixed/only.txt", body=b"ok") headers = signer("PUT", "/mixed/only.txt", body=b"ok")
assert client.put("/mixed/only.txt", headers=headers, data=b"ok").status_code == 200 assert client.put("/mixed/only.txt", headers=headers, data=b"ok").status_code == 200
@@ -270,10 +251,9 @@ def test_principal_dict_with_object_get_only(client, signer):
}, },
], ],
} }
import json
policy_bytes = json.dumps(policy).encode("utf-8") policy_bytes = json.dumps(policy).encode("utf-8")
headers = signer("PUT", "/bucket-policy/mixed", headers={"Content-Type": "application/json"}, body=policy_bytes) headers = signer("PUT", "/mixed?policy", headers={"Content-Type": "application/json"}, body=policy_bytes)
assert client.put("/bucket-policy/mixed", headers=headers, json=policy).status_code == 204 assert client.put("/mixed?policy", headers=headers, json=policy).status_code == 204
assert client.get("/mixed").status_code == 403 assert client.get("/mixed").status_code == 403
allowed = client.get("/mixed/only.txt") allowed = client.get("/mixed/only.txt")
@@ -282,18 +262,20 @@ def test_principal_dict_with_object_get_only(client, signer):
headers = signer("DELETE", "/mixed/only.txt") headers = signer("DELETE", "/mixed/only.txt")
assert client.delete("/mixed/only.txt", headers=headers).status_code == 204 assert client.delete("/mixed/only.txt", headers=headers).status_code == 204
headers = signer("DELETE", "/bucket-policy/mixed") headers = signer("DELETE", "/mixed?policy")
assert client.delete("/bucket-policy/mixed", headers=headers).status_code == 204 assert client.delete("/mixed?policy", headers=headers).status_code == 204
headers = signer("DELETE", "/mixed") headers = signer("DELETE", "/mixed")
assert client.delete("/mixed", headers=headers).status_code == 204 assert client.delete("/mixed", headers=headers).status_code == 204
def test_bucket_policy_wildcard_resource_allows_object_get(client, signer): def test_bucket_policy_wildcard_resource_allows_object_get(client, signer):
import json
headers = signer("PUT", "/test") headers = signer("PUT", "/test")
assert client.put("/test", headers=headers).status_code == 200 assert client.put("/test", headers=headers).status_code == 200
headers = signer("PUT", "/test/vid.mp4", body=b"video") headers = signer("PUT", "/test/vid.mp4", body=b"video")
assert client.put("/test/vid.mp4", headers=headers, data=b"video").status_code == 200 assert client.put("/test/vid.mp4", headers=headers, data=b"video").status_code == 200
@@ -314,10 +296,9 @@ def test_bucket_policy_wildcard_resource_allows_object_get(client, signer):
}, },
], ],
} }
import json
policy_bytes = json.dumps(policy).encode("utf-8") policy_bytes = json.dumps(policy).encode("utf-8")
headers = signer("PUT", "/bucket-policy/test", headers={"Content-Type": "application/json"}, body=policy_bytes) headers = signer("PUT", "/test?policy", headers={"Content-Type": "application/json"}, body=policy_bytes)
assert client.put("/bucket-policy/test", headers=headers, json=policy).status_code == 204 assert client.put("/test?policy", headers=headers, json=policy).status_code == 204
listing = client.get("/test") listing = client.get("/test")
assert listing.status_code == 403 assert listing.status_code == 403
@@ -327,10 +308,10 @@ def test_bucket_policy_wildcard_resource_allows_object_get(client, signer):
headers = signer("DELETE", "/test/vid.mp4") headers = signer("DELETE", "/test/vid.mp4")
assert client.delete("/test/vid.mp4", headers=headers).status_code == 204 assert client.delete("/test/vid.mp4", headers=headers).status_code == 204
headers = signer("DELETE", "/bucket-policy/test") headers = signer("DELETE", "/test?policy")
assert client.delete("/bucket-policy/test", headers=headers).status_code == 204 assert client.delete("/test?policy", headers=headers).status_code == 204
headers = signer("DELETE", "/test") headers = signer("DELETE", "/test")
assert client.delete("/test", headers=headers).status_code == 204 assert client.delete("/test", headers=headers).status_code == 204