stackchain-dashboard/tests/test_issue_attachments.py

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": "![checkout.png](<https://forge.example/attachments/checkout.png>)",
    }
    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": "http://127.0.0.1:3000/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": "http://127.0.0.1:3000/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"] == (
        "![screen\\](not-an-image).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"] == (
        "![pull.png](<https://forge.example/attachments/pull.png>)"
    )
    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": "http://127.0.0.1:3000/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"] == "http://127.0.0.1:3000/attachments/pull.png"