535 lines
19 KiB
Python
535 lines
19 KiB
Python
import asyncio
|
|
import base64
|
|
import io
|
|
import time
|
|
|
|
import httpx
|
|
import pytest
|
|
from PIL import Image, PngImagePlugin
|
|
|
|
from src import gitea_proxy, main
|
|
|
|
|
|
def make_png(color="red", size=(2, 2)):
|
|
output = io.BytesIO()
|
|
Image.new("RGB", size, color).save(output, format="PNG")
|
|
return output.getvalue()
|
|
|
|
|
|
PNG_BYTES = make_png()
|
|
|
|
|
|
def png_with_private_metadata():
|
|
output = io.BytesIO()
|
|
metadata = PngImagePlugin.PngInfo()
|
|
metadata.add_text("Location", "private coordinates")
|
|
Image.new("RGB", (2, 2), "red").save(output, format="PNG", pnginfo=metadata)
|
|
return output.getvalue()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_sanitization_keeps_event_loop_responsive(monkeypatch):
|
|
def slow_sanitizer(_content_type, content):
|
|
time.sleep(0.1)
|
|
return content
|
|
|
|
monkeypatch.setattr(main, "sanitize_image", slow_sanitizer)
|
|
started = time.monotonic()
|
|
ticks = []
|
|
|
|
async def tick():
|
|
await asyncio.sleep(0.01)
|
|
ticks.append(time.monotonic() - started)
|
|
|
|
await asyncio.gather(main._sanitize_attachment("image/png", b"image"), tick())
|
|
|
|
assert ticks[0] < 0.05
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_attachment_idempotency():
|
|
main._authored_action_operations.clear()
|
|
main._idempotency_ledger.clear()
|
|
yield
|
|
main._authored_action_operations.clear()
|
|
main._idempotency_ledger.clear()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_uploads_valid_screenshot_to_assigned_issue(monkeypatch):
|
|
calls = []
|
|
|
|
async def upload(repository, number, filename, content_type, content):
|
|
calls.append((repository, number, filename, content_type, content))
|
|
return {
|
|
"name": "checkout.png",
|
|
"url": "https://forge.example/attachments/checkout.png",
|
|
"size": len(content),
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "upload_assigned_issue_attachment", upload, raising=False
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json={
|
|
"filename": "checkout.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(PNG_BYTES).decode("ascii"),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {
|
|
"name": "checkout.png",
|
|
"url": "https://forge.example/attachments/checkout.png",
|
|
"size": len(PNG_BYTES),
|
|
"markdown": "",
|
|
}
|
|
assert calls == [
|
|
("stackchain/api", 17, "checkout.png", "image/png", PNG_BYTES)
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_strips_metadata_before_upstream(monkeypatch):
|
|
uploaded = []
|
|
source = png_with_private_metadata()
|
|
|
|
async def upload(_repository, _number, filename, content_type, content):
|
|
uploaded.append(content)
|
|
return {"name": filename, "url": "https://forge.example/a.png", "size": len(content)}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
files={"file": ("evidence.png", source, "image/png")},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert uploaded and uploaded[0] != source
|
|
with Image.open(io.BytesIO(uploaded[0])) as sanitized:
|
|
assert sanitized.info.get("Location") is None
|
|
assert sanitized.getpixel((0, 0)) == (255, 0, 0)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_accepts_binary_multipart_without_base64_expansion(monkeypatch):
|
|
calls = []
|
|
|
|
async def upload(repository, number, filename, content_type, content):
|
|
calls.append((repository, number, filename, content_type, content))
|
|
return {
|
|
"name": filename,
|
|
"url": "https://forge.example/attachments/checkout.png",
|
|
"size": len(content),
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
files={"file": ("checkout.png", PNG_BYTES, "image/png")},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert calls == [
|
|
("stackchain/api", 17, "checkout.png", "image/png", PNG_BYTES)
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_replays_confirmed_upload_for_same_key(monkeypatch):
|
|
calls = []
|
|
|
|
async def upload(repository, number, filename, content_type, content):
|
|
calls.append((repository, number, filename, content_type, content))
|
|
return {
|
|
"name": filename,
|
|
"url": "https://forge.example/attachments/checkout.png",
|
|
"size": len(content),
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
|
|
payload = {
|
|
"filename": "checkout.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(PNG_BYTES).decode("ascii"),
|
|
}
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json=payload,
|
|
headers={"Idempotency-Key": "attachment-retry-471"},
|
|
)
|
|
replay = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json=payload,
|
|
headers={"Idempotency-Key": "attachment-retry-471"},
|
|
)
|
|
|
|
assert first.status_code == replay.status_code == 201
|
|
assert replay.json() == first.json()
|
|
assert len(calls) == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_coalesces_concurrent_upload_retries(monkeypatch):
|
|
calls = 0
|
|
|
|
async def upload(_repository, _number, filename, _content_type, content):
|
|
nonlocal calls
|
|
calls += 1
|
|
await asyncio.sleep(0.05)
|
|
return {
|
|
"name": filename,
|
|
"url": "https://forge.example/attachments/checkout.png",
|
|
"size": len(content),
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
|
|
payload = {
|
|
"filename": "checkout.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(PNG_BYTES).decode("ascii"),
|
|
}
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first, retry = await asyncio.gather(*[
|
|
client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json=payload,
|
|
headers={"Idempotency-Key": "attachment-concurrent-471"},
|
|
)
|
|
for _ in range(2)
|
|
])
|
|
|
|
assert first.status_code == retry.status_code == 201
|
|
assert first.json() == retry.json()
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_rejects_changed_upload_for_used_key(monkeypatch):
|
|
calls = 0
|
|
|
|
async def upload(_repository, _number, filename, _content_type, content):
|
|
nonlocal calls
|
|
calls += 1
|
|
return {
|
|
"name": filename,
|
|
"url": "https://forge.example/attachments/checkout.png",
|
|
"size": len(content),
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
accepted = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json={
|
|
"filename": "checkout.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(PNG_BYTES).decode("ascii"),
|
|
},
|
|
headers={"Idempotency-Key": "attachment-conflict-471"},
|
|
)
|
|
conflict = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json={
|
|
"filename": "checkout.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(make_png("blue")).decode("ascii"),
|
|
},
|
|
headers={"Idempotency-Key": "attachment-conflict-471"},
|
|
)
|
|
|
|
assert accepted.status_code == 201
|
|
assert conflict.status_code == 409
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_admits_a_normal_phone_screenshot(monkeypatch):
|
|
screenshot = make_png(size=(390, 844))
|
|
|
|
async def upload(_repository, _number, filename, _content_type, content):
|
|
return {
|
|
"name": filename,
|
|
"url": "https://forge.example/attachments/screen.png",
|
|
"size": len(content),
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "upload_assigned_issue_attachment", upload, raising=False
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json={
|
|
"filename": "screen.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(screenshot).decode("ascii"),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["size"] == len(screenshot)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_rejects_spoofed_image_before_upstream(monkeypatch):
|
|
called = False
|
|
|
|
async def upload(*_args):
|
|
nonlocal called
|
|
called = True
|
|
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "upload_assigned_issue_attachment", upload, raising=False
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json={
|
|
"filename": "not-really.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(b"<script>alert(1)</script>").decode("ascii"),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_rejects_image_over_two_megabytes_before_upstream(monkeypatch):
|
|
called = False
|
|
|
|
async def upload(*_args):
|
|
nonlocal called
|
|
called = True
|
|
return {"name": "large.png", "url": "https://forge.example/a", "size": 1}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
|
|
oversized = b"\x89PNG\r\n\x1a\n" + b"x" * (2 * 1024 * 1024 - 7)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json={
|
|
"filename": "large.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(oversized).decode("ascii"),
|
|
},
|
|
)
|
|
|
|
assert len(oversized) == 2 * 1024 * 1024 + 1
|
|
assert response.status_code == 422
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_attachment_revalidates_assignment_and_sends_multipart():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.method == "GET":
|
|
return httpx.Response(200, json={
|
|
"state": "open", "pull_request": None,
|
|
"assignees": [{"login": "timmy"}],
|
|
})
|
|
return httpx.Response(201, json={
|
|
"id": 9,
|
|
"name": "checkout.png",
|
|
"size": len(PNG_BYTES),
|
|
"browser_download_url": "https://forge.example/attachments/checkout.png",
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.upload_assigned_issue_attachment(
|
|
"stackchain/api", 17, "checkout.png", "image/png", PNG_BYTES
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert [(request.method, request.url.path) for request in requests] == [
|
|
("GET", "/api/v1/user"),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/17"),
|
|
("POST", "/api/v1/repos/stackchain/api/issues/17/assets"),
|
|
]
|
|
upload = requests[-1]
|
|
assert upload.url.params["name"] == "checkout.png"
|
|
assert "multipart/form-data" in upload.headers["content-type"]
|
|
assert b'name="attachment"; filename="checkout.png"' in upload.content
|
|
assert PNG_BYTES in upload.content
|
|
assert result == {
|
|
"name": "checkout.png",
|
|
"url": "https://forge.example/attachments/checkout.png",
|
|
"size": len(PNG_BYTES),
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_markdown_escapes_untrusted_confirmed_filename(monkeypatch):
|
|
async def upload(*_args):
|
|
return {
|
|
"name": "screen](not-an-image).png",
|
|
"url": "https://forge.example/attachments/a.png",
|
|
"size": len(PNG_BYTES),
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json={
|
|
"filename": "screen.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(PNG_BYTES).decode("ascii"),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["markdown"] == (
|
|
".png](<https://forge.example/attachments/a.png>)"
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_hides_an_issue_that_is_no_longer_assigned(monkeypatch):
|
|
async def upload(*_args):
|
|
raise gitea_proxy.IssueNotAvailableError("not assigned")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json={
|
|
"filename": "screen.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(PNG_BYTES).decode("ascii"),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_attachment_endpoint_rejects_unsafe_or_mismatched_filename(monkeypatch):
|
|
called = False
|
|
|
|
async def upload(*_args):
|
|
nonlocal called
|
|
called = True
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
traversal = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json={
|
|
"filename": "../payload.png",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(PNG_BYTES).decode("ascii"),
|
|
},
|
|
)
|
|
mismatch = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/17/attachments",
|
|
json={
|
|
"filename": "payload.html",
|
|
"content_type": "image/png",
|
|
"data": base64.b64encode(PNG_BYTES).decode("ascii"),
|
|
},
|
|
)
|
|
|
|
assert [traversal.status_code, mismatch.status_code] == [422, 422]
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pull_attachment_endpoint_uploads_binary_to_exact_assigned_pull(monkeypatch):
|
|
calls = []
|
|
|
|
async def upload(repository, number, filename, content_type, content):
|
|
calls.append((repository, number, filename, content_type, content))
|
|
return {
|
|
"name": filename,
|
|
"url": "https://forge.example/attachments/pull.png",
|
|
"size": len(content),
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "upload_assigned_pull_attachment", upload, raising=False
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/web/pulls/31/attachments",
|
|
files={"file": ("pull.png", PNG_BYTES, "image/png")},
|
|
headers={"Idempotency-Key": "pull-image-31:attachment"},
|
|
)
|
|
replay = await client.post(
|
|
"/api/v1/repos/stackchain/web/pulls/31/attachments",
|
|
files={"file": ("pull.png", PNG_BYTES, "image/png")},
|
|
headers={"Idempotency-Key": "pull-image-31:attachment"},
|
|
)
|
|
|
|
assert response.status_code == replay.status_code == 201
|
|
assert replay.json() == response.json()
|
|
assert response.json()["markdown"] == (
|
|
""
|
|
)
|
|
assert calls == [("stackchain/web", 31, "pull.png", "image/png", PNG_BYTES)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_pull_attachment_revalidates_assignment_before_exact_pull_upload():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.method == "GET":
|
|
return httpx.Response(200, json={
|
|
"state": "open", "assignees": [{"login": "timmy"}],
|
|
})
|
|
return httpx.Response(201, json={
|
|
"name": "pull.png", "size": len(PNG_BYTES),
|
|
"browser_download_url": "https://forge.example/attachments/pull.png",
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.upload_assigned_pull_attachment(
|
|
"stackchain/web", 31, "pull.png", "image/png", PNG_BYTES
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert [(request.method, request.url.path) for request in requests] == [
|
|
("GET", "/api/v1/user"),
|
|
("GET", "/api/v1/repos/stackchain/web/pulls/31"),
|
|
("POST", "/api/v1/repos/stackchain/web/issues/31/assets"),
|
|
]
|
|
assert PNG_BYTES in requests[-1].content
|
|
assert result["url"] == "https://forge.example/attachments/pull.png"
|