86 lines
3.3 KiB
Python
86 lines
3.3 KiB
Python
import os
|
|
import pytest
|
|
import hashlib
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from app.main import app
|
|
from app.core.database import init_db, AsyncSessionLocal
|
|
from app.models.models import Client, ClientCredential
|
|
from app.core.security import hash_token
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_isolation_security():
|
|
"""
|
|
Ensures that Client A cannot access, query, upload to, or complete sessions
|
|
belonging to Client B.
|
|
"""
|
|
await init_db()
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
# Create Client A
|
|
async with AsyncSessionLocal() as db:
|
|
client_a = Client(
|
|
client_code="CLIENT-ISO-A",
|
|
name="Company A WinServer",
|
|
status="ONLINE"
|
|
)
|
|
db.add(client_a)
|
|
await db.flush()
|
|
token_a = "token-secret-client-a"
|
|
cred_a = ClientCredential(
|
|
client_id=client_a.id,
|
|
device_id="dev-iso-a",
|
|
token_hash=hash_token(token_a),
|
|
name="Cred A"
|
|
)
|
|
db.add(cred_a)
|
|
|
|
# Create Client B
|
|
client_b = Client(
|
|
client_code="CLIENT-ISO-B",
|
|
name="Company B WinServer",
|
|
status="ONLINE"
|
|
)
|
|
db.add(client_b)
|
|
await db.flush()
|
|
token_b = "token-secret-client-b"
|
|
cred_b = ClientCredential(
|
|
client_id=client_b.id,
|
|
device_id="dev-iso-b",
|
|
token_hash=hash_token(token_b),
|
|
name="Cred B"
|
|
)
|
|
db.add(cred_b)
|
|
await db.commit()
|
|
|
|
headers_a = {"X-Device-Id": "dev-iso-a", "X-Device-Token": token_a}
|
|
headers_b = {"X-Device-Id": "dev-iso-b", "X-Device-Token": token_b}
|
|
|
|
# Client A starts an upload session
|
|
resp_a = await ac.post("/api/upload/session", json={
|
|
"filename": "confidential_a.bak",
|
|
"file_size": 1024 * 1024,
|
|
"sha256": hashlib.sha256(b"secret_a_data").hexdigest(),
|
|
"chunk_size": 1024 * 1024
|
|
}, headers=headers_a)
|
|
assert resp_a.status_code == 200
|
|
session_a_code = resp_a.json()["session_code"]
|
|
|
|
# Client B attempts to view or hijack Client A's session -> MUST be denied (404 / 403)
|
|
hijack_status = await ac.get(f"/api/upload/{session_a_code}/status", headers=headers_b)
|
|
assert hijack_status.status_code == 404, "Security violation: Client B accessed Client A session!"
|
|
|
|
# Client B attempts to upload chunk to Client A's session -> MUST be denied
|
|
hijack_chunk = await ac.post(
|
|
f"/api/upload/{session_a_code}/chunk?chunk_index=0",
|
|
content=b"malicious_bytes",
|
|
headers={**headers_b, "Content-Type": "application/octet-stream"}
|
|
)
|
|
assert hijack_chunk.status_code == 404, "Security violation: Client B injected chunk into Client A session!"
|
|
|
|
# Client B attempts to complete Client A's session -> MUST be denied
|
|
hijack_complete = await ac.post(f"/api/upload/{session_a_code}/complete", headers=headers_b)
|
|
assert hijack_complete.status_code == 404, "Security violation: Client B triggered completion on Client A session!"
|
|
print("\n>>> Isolation Test Passed: Strict multitenant isolation verified!")
|