"""Real HTTP/worker/storage and adverse ZIP cases: independent capped VM only.""" import asyncio import hashlib import io import json import os import resource import signal import struct import subprocess import sys import time import zipfile from pathlib import Path from uuid import UUID, uuid4 import httpx import pytest from conftest import connect from sqlalchemy.engine import make_url from kf_oj.contracts.judging import ArtifactReference from kf_oj.contracts.judging_v2 import AdmissionRegistry from kf_oj.infra.artifacts.archive_reader import ArchiveError, ArchiveReader from kf_oj.infra.artifacts.staging import ImportStaging from kf_oj.infra.artifacts.store import ArtifactError, ArtifactStore from kf_oj.modules.identity.passwords import hash_password ROOT = Path(__file__).resolve().parents[2] @pytest.fixture def independent_import_vm(): marker = Path("/kf/INDEPENDENT_VM") if not marker.exists() or marker.read_text().strip() != "isolated-import-test-guest": pytest.skip("adverse archive/full import acceptance requires the independent capped VM") assert [p.name for p in Path("/sys/class/net").iterdir()] == ["lo"] assert all( x not in Path("/proc/self/mountinfo").read_text() for x in ("9p", "virtiofs", "hgfs") ) return {name: Path("/sys/fs/cgroup/kf-total") / name for name in ("api", "import-worker")} def package( policy, *, multi=False, method=zipfile.ZIP_STORED, mutation=None, payload_mutation=None ): payload = { "statement.md": b"Public statement only", "hidden.in": b"synthetic-hidden-input\x00\xff", "hidden.out": b"synthetic-hidden-output\x00\xfe", "sample.in": b"1\n", "sample.out": b"2\n", } manifest = { "schema_version": 1, "slug": "web-synthetic", "title": "Public", "statement_file": "statement.md", "languages": ["c17", "cpp20"], "toolchain": policy.legacy.toolchain.model_dump(mode="json"), "comparison": "exact_bytes", "compile_limits": policy.legacy.compile_ceiling.model_dump(), "run_limits": policy.legacy.run_ceiling.model_dump(), "job_wall_time_us": 60000000, "cases": [{"key": "binary", "input_file": "hidden.in", "expected_file": "hidden.out"}], "samples": [{"input_file": "sample.in", "output_file": "sample.out"}], } if multi: for key in ("languages", "toolchain", "compile_limits", "run_limits", "job_wall_time_us"): del manifest[key] manifest["schema_version"] = 2 manifest["language_configurations"] = [ { key: json.loads( ( ROOT / f"tests/fixtures/contracts-v2/judge-manifest.valid-{language}.json" ).read_text() )[key] for key in ( "language", "toolchain", "compile_limits", "run_limits", "job_wall_time_us", ) } for language in ("c17", "cpp20", "python313", "java21") ] if mutation: mutation(manifest) payload["problem.json"] = json.dumps(manifest).encode() if payload_mutation: payload_mutation(payload) out = io.BytesIO() with zipfile.ZipFile(out, "w", compression=method) as archive: for name, data in payload.items(): archive.writestr(name, data) return out.getvalue(), payload def test_bounded_zip_and_streaming_adoption(independent_import_vm, tmp_path): policy = AdmissionRegistry.model_validate_json( (ROOT / "tests/fixtures/contracts-v2/judge-registry.valid.json").read_bytes() ) for method in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED): blob, payload = package( policy, method=method, payload_mutation=lambda p: p.update({"hidden.in": b"A" * 1048576 + b"\x00\xff"}), ) file = tmp_path / f"archive-{method}" file.write_bytes(blob) file.chmod(0o600) output = tmp_path / f"members-{method}" output.mkdir(mode=0o700) fd = os.open(file, os.O_RDONLY) directory = os.open(output, os.O_RDONLY | os.O_DIRECTORY) try: reader = ArchiveReader(fd, len(blob), deadline=time.monotonic() + 60) reader.verify_digest(hashlib.sha256(blob).hexdigest()) assert reader.read("problem.json", 262144) == payload.pop("problem.json") references = reader.extract_remaining(directory) assert reader.unpacked_bytes == sum(len(v) for v in payload.values()) + len( package(policy, method=method)[1]["problem.json"] ) store = ArtifactStore(tmp_path / f"objects-{method}") try: for name, (digest, size) in references.items(): assert (output / name).read_bytes() == payload[name] source = os.open(name, os.O_RDONLY, dir_fd=directory) try: reference = ArtifactReference(digest=digest, bytes=size) assert store.put_file(source, reference) == reference assert store.put_file(source, reference) == reference finally: os.close(source) finally: store.close() finally: os.close(fd) os.close(directory) good, _ = package(policy) wrong_crc = bytearray(good) struct.pack_into("clock_timestamp() " "FROM problem_import_pool WHERE id=1" ): break except httpx.ConnectError: pass await asyncio.sleep(0.1) else: raise AssertionError( "real API/worker startup deadline; api=" + (tmp_path / "api.log").read_text()[-1500:] + " worker=" + (tmp_path / "worker.log").read_text()[-1500:] ) prefix = "/api/v1/admin/problem-imports" assert (await client.get(prefix + "/capabilities")).status_code == 401 async def login(name): client.cookies.clear() r = await client.post( "/api/v1/auth/login", json={"username": name, "password": password}, headers={"Origin": "http://127.0.0.1:58000"}, ) assert r.status_code == 200, r.text return { "Origin": "http://127.0.0.1:58000", "X-CSRF-Token": client.cookies["kf_csrf"], } student_headers = await login("web_student") assert (await client.get(prefix + "/capabilities")).status_code == 403 assert ( await client.post( prefix, json={ "operation_id": str(uuid4()), "archive_digest": "a" * 64, "archive_bytes": 100, }, headers=student_headers, ) ).status_code == 403 headers = await login("web_admin") assert (await client.get(prefix + "/capabilities")).json()[ "uploads_available" ] is True for invalid in ( {}, {"Origin": "http://127.0.0.1:58000", "X-CSRF-Token": "0" * 64}, headers | {"Origin": "https://evil.invalid"}, headers | {"Origin": "https://evil.invalid", "X-Forwarded-Host": "127.0.0.1:58000"}, ): assert ( await client.post( prefix, json={ "operation_id": str(uuid4()), "archive_digest": "a" * 64, "archive_bytes": 100, }, headers=invalid, ) ).status_code == 403 async def create(blob, key=None): key = key or uuid4() body = { "operation_id": str(key), "archive_digest": hashlib.sha256(blob).hexdigest(), "archive_bytes": len(blob), } deadline = time.monotonic() + 15 while True: r = await client.post(prefix, json=body, headers=headers) if r.status_code != 503 or time.monotonic() >= deadline: break await asyncio.sleep(0.2) assert r.status_code == 201, r.text assert r.headers["cache-control"] == "no-store" assert ( await client.post(prefix, json=body, headers=headers) ).json() == r.json() assert ( await client.post( prefix, json=body | {"archive_digest": "f" * 64}, headers=headers ) ).status_code == 409 return str(key) async def put_blob(blob, key): deadline = time.monotonic() + 30 while True: result = await client.put( f"{prefix}/{key}/archive", content=blob, headers=headers | {"Content-Type": "application/zip"}, ) if result.status_code != 503 or time.monotonic() >= deadline: return result await asyncio.sleep(0.2) async def discard_owned(key): deadline = time.monotonic() + 15 while True: result = await client.post(f"{prefix}/{key}/discard", headers=headers) if result.status_code != 503 or time.monotonic() >= deadline: return result await asyncio.sleep(0.2) async def upload(blob, key): r = await put_blob(blob, key) assert r.status_code == 200, r.text deadline = time.monotonic() + 100 while time.monotonic() < deadline: r = await client.get(f"{prefix}/{key}") assert r.status_code == 200, r.text if r.json()["state"] in ("ready", "failed"): return r.json() assert processes["worker"].poll() is None, ( tmp_path / "worker.log" ).read_text()[-2000:] await asyncio.sleep(0.2) raise AssertionError( "candidate deadline; " + (tmp_path / "worker.log").read_text()[-2000:] ) async def wait_released(key, state, code=None, seconds=120): deadline = time.monotonic() + seconds while time.monotonic() < deadline: row = await c.fetchrow( "SELECT state,failure_code,reserved_bytes,attempts FROM " "problem_import_operations WHERE id=$1", UUID(key), ) if row["state"] == state and row["reserved_bytes"] == 0: if code: assert row["failure_code"] == code, dict(row) return row await asyncio.sleep(0.1) raise AssertionError("operation cleanup deadline: " + str(dict(row))) async def restart_ready_worker(): previous = await c.fetchval( "SELECT worker_ready_until FROM problem_import_pool WHERE id=1" ) assert previous is not None start("worker") deadline = time.monotonic() + 300 while time.monotonic() < deadline: assert processes["worker"].poll() is None if await c.fetchval( "SELECT worker_ready_until>clock_timestamp()+interval '80 seconds' " "AND worker_ready_until>$1 FROM problem_import_pool WHERE id=1", previous, ): return await asyncio.sleep(0.1) raise AssertionError("fresh worker heartbeat deadline") async def raw_body(key, declared, data, *, delay=0, disconnect=False): reader, writer = await asyncio.open_connection("127.0.0.1", 58000) cookie = "; ".join(f"{k}={v}" for k, v in client.cookies.items()) request = ( f"PUT {prefix}/{key}/archive HTTP/1.1\r\n" "Host: 127.0.0.1:58000\r\nOrigin: http://127.0.0.1:58000\r\n" f"Cookie: {cookie}\r\nX-CSRF-Token: {headers['X-CSRF-Token']}\r\n" f"Content-Type: application/zip\r\nContent-Length: {declared}\r\n" "Connection: close\r\n\r\n" ).encode() try: writer.write(request + data) await writer.drain() if delay: await asyncio.sleep(delay) if disconnect: return None return await asyncio.wait_for(reader.read(8192), 45) finally: writer.close() await writer.wait_closed() if not recovery_only: for multi, method in ( (False, zipfile.ZIP_STORED), (True, zipfile.ZIP_DEFLATED), ): blob, payload = package(policy, multi=multi, method=method) key = await create(blob) detail = await upload(blob, key) assert detail["state"] == "ready", (detail["failure_code"], detail["state"]) print( "ACTUAL_READY_IMPORT", multi, method, detail["candidate_id"], flush=True ) await wait_released(key, "ready") retry = await put_blob(blob, key) assert retry.status_code == 200 different = blob[:-1] + bytes([blob[-1] ^ 1]) conflict = await put_blob(different, key) assert conflict.status_code == 409 assert (await client.get(f"{prefix}/{key}")).json()[ "candidate_id" ] == detail["candidate_id"] assert detail["candidate"]["judging_schema_version"] == (2 if multi else 1) assert detail["candidate"]["case_count"] == 1 assert len(detail["candidate"]["languages"]) == (4 if multi else 2) assert "synthetic-hidden" not in json.dumps(detail) and str( tmp_path ) not in json.dumps(detail) # Actual database enforcement, not trigger inspection alone. worker_db = await connect( make_url(postgres_url) .set(username=roles["worker"]) .render_as_string(hide_password=False) ) try: from asyncpg import CheckViolationError, InsufficientPrivilegeError with pytest.raises(InsufficientPrivilegeError): await worker_db.execute( "UPDATE problem_candidates SET preview=preview WHERE id=$1", UUID(detail["candidate_id"]), ) with pytest.raises(CheckViolationError): await worker_db.execute( "DELETE FROM problem_candidates WHERE id=$1", UUID(detail["candidate_id"]), ) with pytest.raises(CheckViolationError): await worker_db.execute( """INSERT INTO problem_candidate_cases (candidate_id,position,key,input_digest,input_bytes,expected_digest,expected_bytes) SELECT candidate_id,1,'late',input_digest,input_bytes, expected_digest,expected_bytes FROM problem_candidate_cases WHERE candidate_id=$1""", UUID(detail["candidate_id"]), ) finally: await worker_db.close() refs = await c.fetch( "SELECT input_digest,input_bytes,expected_digest,expected_bytes " "FROM problem_candidate_cases WHERE candidate_id=$1", UUID(detail["candidate_id"]), ) assert len(refs) == 1 for field, name in (("input", "hidden.in"), ("expected", "hidden.out")): digest = refs[0][field + "_digest"] data = (artifact / digest).read_bytes() assert ( data == payload[name] and hashlib.sha256(data).hexdigest() == digest ) assert await c.fetchval("SELECT count(*) FROM problem_versions") == 0 assert (await client.get("/api/v1/problems")).json()["items"] == [] assert ( await c.fetchval( "SELECT reserved_bytes FROM problem_import_pool WHERE id=1" ) == 0 ) assert sorted(p.name for p in staging.iterdir()) == [".lock"] def reuse_public_member(payload): declaration = json.loads(payload["problem.json"]) declaration["samples"][0]["output_file"] = "sample.in" payload["problem.json"] = json.dumps(declaration).encode() del payload["sample.out"] # Actual child consumes malformed data but never executes contents. for mutation in ( lambda p: p.update({"../escape": b"bad"}), lambda p: p.update({"unknown.bin": b"bad"}), lambda p: p.update({"hidden.in": b"PK\x03\x04" + b"bad"}), lambda p: p.update({"sample.in": p["hidden.in"]}), reuse_public_member, ): blob, _ = package(policy, payload_mutation=mutation) key = await create(blob) result = await upload(blob, key) # Publicly declared samples can happen to equal hidden bytes; declarations, # not byte equality, decide visibility. This last finite case is legal. if ( b"PK\x03\x04bad" in blob or b"../escape" in blob or b"unknown.bin" in blob ): assert result["state"] == "failed" and result["candidate"] is None, ( result ) else: assert result["state"] == "ready", result await wait_released(key, result["state"]) assert ( await c.fetchval( "SELECT reserved_bytes FROM problem_import_pool WHERE id=1" ) == 0 ) blob, _ = package(policy) key = await create(blob) stop("worker") occupied = ImportStaging(staging) try: with occupied.exclusive(): mismatch = await client.put( f"{prefix}/{key}/archive", content=blob[:-1], headers=headers | {"Content-Type": "application/zip"}, ) assert mismatch.status_code == 409 finally: occupied.close() assert (await discard_owned(key)).status_code == 200 print("ACTUAL_IMPORT_CANDIDATE_BOUNDARIES_ACCEPTED", flush=True) return blob, _ = package(policy) # Restart after receipt before consumption preserves the immutable operation. stop("worker") key = await create(blob) r = await client.put( f"{prefix}/{key}/archive", content=blob, headers=headers | {"Content-Type": "application/zip"}, ) assert r.status_code == 200 and r.json()["state"] == "uploaded" start("worker") began = time.monotonic() deadline = began + 300 previous_state = None while time.monotonic() < deadline: detail = (await client.get(f"{prefix}/{key}")).json() current_state = detail["state"] if current_state != previous_state: print( "ACTUAL_RESTART_OBSERVATION", round(time.monotonic() - began, 3), current_state, flush=True, ) previous_state = current_state if detail["state"] in ("ready", "failed"): break await asyncio.sleep(1) assert detail["state"] == "ready", (detail["failure_code"], detail["state"]) assert ( await c.fetchval( "SELECT attempts FROM problem_import_operations WHERE id=$1", UUID(key), ) == 1 ) await wait_released(key, "ready") # A second real administrator cannot inspect, upload or discard another's candidate. headers = await login("web_backup") stop("worker") occupied = ImportStaging(staging) try: with occupied.exclusive(): assert (await client.get(f"{prefix}/{key}")).status_code == 404 assert ( await client.post(f"{prefix}/{key}/discard", headers=headers) ).status_code == 404 assert ( await client.put( f"{prefix}/{key}/archive", content=blob, headers=headers | {"Content-Type": "application/zip"}, ) ).status_code == 404 finally: occupied.close() start("worker") headers = await login("web_admin") # Incomplete TCP body and actual 30-second streaming deadline free reservations. interrupted = await create(blob) stop("worker") # Exercise the upload path without a competing GC root lock. await raw_body(interrupted, len(blob), blob[:8], disconnect=True) await wait_released(interrupted, "failed", "upload_interrupted") await restart_ready_worker() slow = await create(blob) stop("worker") response = await raw_body(slow, len(blob), blob[:8], delay=33) await wait_released(slow, "failed", "upload_deadline") assert response.startswith(b"HTTP/1.1 408"), response[:200] await restart_ready_worker() # A full received body with the wrong SHA is never acknowledged as an upload. wrong = await create(blob) stop("worker") r = await client.put( f"{prefix}/{wrong}/archive", content=blob[:-1] + b"!", headers=headers | {"Content-Type": "application/zip"}, ) assert r.status_code == 409 await wait_released(wrong, "failed", "digest_mismatch") await restart_ready_worker() # Two actual reservations fill the global staging pool with Worker stopped. stop("worker") queued = [] for _ in range(2): queued_key = await create(blob) r = await client.put( f"{prefix}/{queued_key}/archive", content=blob, headers=headers | {"Content-Type": "application/zip"}, ) assert r.status_code == 200 queued.append(queued_key) assert ( await c.fetchval("SELECT reserved_bytes FROM problem_import_pool WHERE id=1") == 268435456 ) r = await client.post( prefix, json={ "operation_id": str(uuid4()), "archive_digest": hashlib.sha256(blob).hexdigest(), "archive_bytes": len(blob), }, headers=headers, ) assert r.status_code == 503 for queued_key in queued: assert ( await client.post(f"{prefix}/{queued_key}/discard", headers=headers) ).status_code == 200 assert ( await c.fetchval("SELECT reserved_bytes FROM problem_import_pool WHERE id=1") == 0 ) start("worker") # Kill parent AND parser in their already verified cgroup while a lease is held. # This finite injection tests both retries. No uploaded code is executed. interrupted_parser = await create(blob) r = await put_blob(blob, interrupted_parser) assert r.status_code == 200 for expected_attempt in (1, 2): deadline = time.monotonic() + 120 while time.monotonic() < deadline: row = await c.fetchrow( "SELECT state,attempts FROM problem_import_operations WHERE id=$1", UUID(interrupted_parser), ) parser = [] for pid in ( groups["import-worker"].joinpath("cgroup.procs").read_text().split() ): cmdline = Path(f"/proc/{pid}/cmdline") if ( cmdline.exists() and b"kf_oj.workers._archive_child" in cmdline.read_bytes() ): parser.append(int(pid)) if ( row["state"] == "validating" and row["attempts"] == expected_attempt and parser ): # Freeze parser, kill parent, then parser. A live child keeps the flock. for pid in parser: os.kill(pid, signal.SIGSTOP) processes["worker"].kill() processes["worker"].wait(timeout=10) await c.execute( "UPDATE problem_import_operations SET lease_until=" "clock_timestamp()-interval '1 second' WHERE id=$1", UUID(interrupted_parser), ) # A live stopped child retains the real kernel lock after parent death. probe = ImportStaging(staging) try: with pytest.raises(ArtifactError): with probe.exclusive(): pass finally: probe.close() denied = await client.post( prefix, json={ "operation_id": str(uuid4()), "archive_digest": hashlib.sha256(blob).hexdigest(), "archive_bytes": len(blob), }, headers=headers, ) assert denied.status_code == 503 for pid in parser: os.kill(pid, signal.SIGKILL) stop("worker") break await asyncio.sleep(0.02) else: raise AssertionError("claimed parser injection deadline") # Advance the lease under the VM test operator; requests cannot set clocks. await c.execute( "UPDATE problem_import_operations SET lease_until=" "clock_timestamp()-interval '1 second' WHERE id=$1", UUID(interrupted_parser), ) start("worker") row = await wait_released(interrupted_parser, "failed", "validation_unavailable") assert row["attempts"] == 2 assert ( await c.fetchval( "SELECT count(*) FROM problem_candidates WHERE operation_id=$1", UUID(interrupted_parser), ) == 0 ) # Real concurrent claims share the same capped process and use separate lock FDs. stop("worker") paired_key = await create(blob) r = await client.put( f"{prefix}/{paired_key}/archive", content=blob, headers=headers | {"Content-Type": "application/zip"}, ) assert r.status_code == 200 and r.json()["state"] == "uploaded" start("worker", pair=True) row = await wait_released(paired_key, "ready") assert row["attempts"] == 1 assert ( await c.fetchval( "SELECT count(*) FROM problem_candidates WHERE operation_id=$1", UUID(paired_key), ) == 1 ) deadline = time.monotonic() + 120 while processes["worker"].poll() is None and time.monotonic() < deadline: await asyncio.sleep(0.1) assert processes["worker"].returncode == 0, (tmp_path / "worker.log").read_text()[ -1500: ] assert ( "ACTUAL_CONCURRENT_IMPORT_CLAIM_ACCEPTED" in (tmp_path / "worker.log").read_text() ) stop("worker") start("worker") # Insert an old receipt with the test operator; immutable timestamps stay unchanged. expired_id = uuid4() async with c.transaction(): await c.execute( """INSERT INTO problem_import_operations (id,owner_id,authorization_version,archive_digest,archive_bytes,created_at, updated_at,expires_at,lease_until,attempts,fence,reserved_bytes) SELECT $1,$2,authorization_version,$3,$4, clock_timestamp()-interval '2 days', clock_timestamp()-interval '2 days',clock_timestamp()-interval '1 day', clock_timestamp()-interval '1 day',0,0,134217728 FROM accounts WHERE id=$2""", expired_id, users["web_admin"], hashlib.sha256(blob).hexdigest(), len(blob), ) assert ( await c.fetchval( "SELECT state FROM problem_import_operations WHERE id=$1", expired_id ) == "uploading" ) await c.execute( "UPDATE problem_import_pool SET reserved_bytes=" "reserved_bytes+134217728 WHERE id=1" ) await wait_released(str(expired_id), "expired", "expired") print( "ACTUAL_IMPORT_DISCONNECT_DEADLINE_CAPACITY_TWO_RETRIES_EXPIRY_ACCEPTED", flush=True, ) # Revocation is live and old cookies cannot read or write management APIs. await c.execute("SELECT * FROM kf_manage_account_authority('web_admin','revoke')") assert (await client.get(prefix + "/capabilities")).status_code == 401 assert ( await client.post(f"{prefix}/{key}/discard", headers=headers) ).status_code == 401 await c.execute("SELECT * FROM kf_manage_account_authority('web_admin','grant')") headers = await login("web_admin") stop("worker") discard_after = await c.fetchval("SELECT clock_timestamp()") page = (await client.get(prefix, params={"limit": 100})).json() for entry in page["items"]: assert ( await client.post( f"{prefix}/{entry['operation_id']}/discard", headers=headers ) ).status_code == 200 await restart_ready_worker() deadline = time.monotonic() + 30 while time.monotonic() < deadline and await c.fetchval( "SELECT count(*) FROM problem_candidates" ): await asyncio.sleep(0.1) assert await c.fetchval("SELECT count(*) FROM problem_candidates") == 0 assert await c.fetchval("SELECT count(*) FROM problem_candidate_cases") == 0 assert ( await c.fetchval("SELECT reserved_bytes FROM problem_import_pool WHERE id=1") == 0 ) assert await c.fetchval("SELECT count(*) FROM problem_maintenance_events") >= 10 assert ( await c.fetchval( "SELECT bool_and(e.authorization_version=a.authorization_version) " "FROM problem_maintenance_events e JOIN accounts a ON a.id=e.actor_id " "WHERE e.action='import_discarded' AND e.actor_id=$1 " "AND e.created_at >= $2", users["web_admin"], discard_after, ) is True ) for table in ( "evaluations", "judge_attempts", "evaluation_performance", "accounts", "sessions", ): for privilege in ("SELECT", "INSERT", "UPDATE", "DELETE"): assert not await c.fetchval( "SELECT has_table_privilege($1,$2,$3)", roles["worker"], table, privilege, ) assert not await c.fetchval( "SELECT has_table_privilege($1,'problem_maintenance_events','UPDATE')", roles["api"], ) for privilege in ("SELECT", "UPDATE", "DELETE"): assert not await c.fetchval( "SELECT has_table_privilege($1,'problem_maintenance_events',$2)", roles["worker"], privilege, ) # Missing configuration is observed through the real server, never preview data. stop("api") start("api", configured=False) deadline = time.monotonic() + 300 while time.monotonic() < deadline: try: r = await client.get(prefix + "/capabilities") if r.status_code == 200: break except httpx.ConnectError: pass await asyncio.sleep(0.1) assert r.status_code == 200 and not r.json()["uploads_available"] assert not r.json()["validation_available"] assert ( await client.post( prefix, json={ "operation_id": str(uuid4()), "archive_digest": hashlib.sha256(blob).hexdigest(), "archive_bytes": len(blob), }, headers=headers, ) ).status_code == 503 print("ACTUAL_IMPORT_HTTP_WORKER_ACCEPTED", flush=True) finally: stop("api") stop("worker") for role in roles.values(): await c.execute(f"DROP OWNED BY {role}") await c.execute(f"DROP ROLE {role}") await c.close() asyncio.run(run())