ctfy.sdk.admin_resources.authoring_review
One competition's uploads, for the people allowed to rule on them.
docs/community-authoring-plan.md P2.3, decision 16. Two resources:
admin.authoring_reviewers— who may rule on one event's uploads (super-admin only), the competition-admin grant's shape.admin.competition_authored_challenges— the review itself, scoped to one event by the path: list, read, audit and rule. The platform queue (admin.authored_challenges) keeps everything else — publish, curate, push, smoke — because what a tree runs stays with a platform admin.
1"""One competition's uploads, for the people allowed to rule on them. 2 3``docs/community-authoring-plan.md`` P2.3, decision 16. Two resources: 4 5* ``admin.authoring_reviewers`` — who may rule on one event's uploads 6 (super-admin only), the competition-admin grant's shape. 7* ``admin.competition_authored_challenges`` — the review itself, scoped 8 to one event by the path: list, read, audit and rule. The platform 9 queue (``admin.authored_challenges``) keeps everything else — publish, 10 curate, push, smoke — because what a tree *runs* stays with a platform 11 admin. 12""" 13 14from __future__ import annotations 15 16import builtins 17 18from ctfy.core.enums import AuthoredDestination, AuthoredQueueOrder, AuthoredSmokeFilter 19from ctfy.core.state.models import ReviewStatus 20from ctfy.sdk._helpers import PagedList, _extract_items, _raise_for_status 21from ctfy.sdk.base import BaseHttpClient 22from ctfy.server.models import ( 23 AdminAuthoredChallengeDetail, 24 AdminAuthoredChallengeRow, 25 AdminAuthoredRevisionRow, 26 AuthoredAuditReport, 27 AuthoredChallengeFileBody, 28 AuthoredRevisionDiff, 29 AuthoringReviewerInfo, 30) 31 32 33class AdminAuthoringReviewersResource: 34 """Grant / revoke the right to rule on one competition's uploads.""" 35 36 def __init__(self, http: BaseHttpClient) -> None: 37 self._http = http 38 39 def list( 40 self, competition_id: str, offset: int = 0, limit: int = 50 41 ) -> PagedList[AuthoringReviewerInfo]: 42 """Who may rule on this competition's uploads.""" 43 resp = self._http.request( 44 "GET", 45 f"/admin/competitions/{competition_id}/authoring-reviewers", 46 params={"offset": offset, "limit": limit}, 47 ) 48 _raise_for_status(resp) 49 return _extract_items(resp.json(), AuthoringReviewerInfo) 50 51 def grant( 52 self, competition_id: str, *, user_id: str = "", email: str = "" 53 ) -> AuthoringReviewerInfo: 54 """Grant it, to a user named by ``user_id`` or ``email``. The 55 holder reaches this event's uploads and nothing else of it — not 56 the organiser's console, not the entrants' registration details.""" 57 resp = self._http.request( 58 "POST", 59 f"/admin/competitions/{competition_id}/authoring-reviewers", 60 json={"user_id": user_id, "email": email}, 61 ) 62 _raise_for_status(resp) 63 return AuthoringReviewerInfo.model_validate(resp.json()) 64 65 def revoke(self, competition_id: str, user_id: str) -> None: 66 """Revoke it.""" 67 resp = self._http.request( 68 "DELETE", f"/admin/competitions/{competition_id}/authoring-reviewers/{user_id}" 69 ) 70 _raise_for_status(resp) 71 72 73class AdminCompetitionAuthoredChallengesResource: 74 """Read and rule on one competition's uploads. 75 76 ⚠️ The competition is the **grant**, not a filter: another event's 77 upload answers 404 here, and there is no parameter that widens the 78 listing. 79 """ 80 81 def __init__(self, http: BaseHttpClient) -> None: 82 self._http = http 83 84 def list( 85 self, 86 competition_id: str, 87 *, 88 review_status: ReviewStatus | None = None, 89 smoke: AuthoredSmokeFilter | None = None, 90 revision: ReviewStatus | None = None, 91 order: AuthoredQueueOrder | None = None, 92 destination: AuthoredDestination | None = None, 93 offset: int = 0, 94 limit: int = 50, 95 ) -> PagedList[AdminAuthoredChallengeRow]: 96 """This event's uploads, oldest first. ``smoke``, ``revision``, 97 ``destination`` and ``order`` work as on the platform queue — except that each 98 uploader's ``author_reputation``, and so the ``reputation`` order, 99 counts only the uploads this reviewer may attribute to them.""" 100 params: dict[str, object] = {"offset": offset, "limit": limit} 101 if review_status is not None: 102 params["review_status"] = review_status 103 if smoke is not None: 104 params["smoke"] = smoke 105 if revision is not None: 106 params["revision"] = revision 107 if order is not None: 108 params["order"] = order 109 if destination is not None: 110 params["destination"] = destination 111 resp = self._http.request( 112 "GET", f"/admin/competitions/{competition_id}/authored-challenges", params=params 113 ) 114 _raise_for_status(resp) 115 return _extract_items(resp.json(), AdminAuthoredChallengeRow) 116 117 def get(self, competition_id: str, challenge_id: str) -> AdminAuthoredChallengeDetail: 118 """One upload, with the paths, sizes and execution surface of 119 what is in it.""" 120 resp = self._http.request( 121 "GET", f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 122 ) 123 _raise_for_status(resp) 124 return AdminAuthoredChallengeDetail.model_validate(resp.json()) 125 126 def audit(self, competition_id: str, challenge_id: str) -> AuthoredAuditReport: 127 """Run the challenge audit over this upload. Read ``outcome`` 128 before the counts — ``timed_out`` and ``failed`` are the audit 129 declining to answer, never a clean bill.""" 130 resp = self._http.request( 131 "GET", 132 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/audit", 133 ) 134 _raise_for_status(resp) 135 return AuthoredAuditReport.model_validate(resp.json()) 136 137 def read_file( 138 self, competition_id: str, challenge_id: str, path: str 139 ) -> AuthoredChallengeFileBody: 140 """One file's contents — attacker-authored text written to be 141 run. ``binary`` and ``truncated`` say what you are not seeing.""" 142 resp = self._http.request( 143 "POST", 144 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/file", 145 json={"path": path}, 146 ) 147 _raise_for_status(resp) 148 return AuthoredChallengeFileBody.model_validate(resp.json()) 149 150 def review( 151 self, 152 competition_id: str, 153 challenge_id: str, 154 *, 155 status: ReviewStatus, 156 note: str = "", 157 ) -> AdminAuthoredChallengeRow: 158 """Record a content verdict. ``rejected`` requires a note, and 159 ``pending`` withdraws a verdict. Publishing stays with a 160 platform admin, who reviews what the tree runs.""" 161 resp = self._http.request( 162 "POST", 163 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/review", 164 json={"status": status, "note": note}, 165 ) 166 _raise_for_status(resp) 167 return AdminAuthoredChallengeRow.model_validate(resp.json()) 168 169 def revisions( 170 self, competition_id: str, challenge_id: str 171 ) -> builtins.list[AdminAuthoredRevisionRow]: 172 """Every revision of one of this event's uploads, newest first.""" 173 resp = self._http.request( 174 "GET", 175 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/revisions", 176 ) 177 _raise_for_status(resp) 178 return [AdminAuthoredRevisionRow.model_validate(row) for row in resp.json()] 179 180 def revision_diff( 181 self, competition_id: str, challenge_id: str, revision_id: str 182 ) -> AuthoredRevisionDiff: 183 """A proposed revision against the live tree (plan P3.1).""" 184 resp = self._http.request( 185 "GET", 186 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 187 f"/revisions/{revision_id}/diff", 188 ) 189 _raise_for_status(resp) 190 return AuthoredRevisionDiff.model_validate(resp.json()) 191 192 def read_revision_file( 193 self, competition_id: str, challenge_id: str, revision_id: str, path: str 194 ) -> AuthoredChallengeFileBody: 195 """One file of a revision, whole.""" 196 resp = self._http.request( 197 "POST", 198 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 199 f"/revisions/{revision_id}/file", 200 json={"path": path}, 201 ) 202 _raise_for_status(resp) 203 return AuthoredChallengeFileBody.model_validate(resp.json()) 204 205 def review_revision( 206 self, 207 competition_id: str, 208 challenge_id: str, 209 revision_id: str, 210 *, 211 status: ReviewStatus, 212 note: str = "", 213 ) -> AdminAuthoredRevisionRow: 214 """A content verdict on a revision of one of this event's uploads. 215 Applying it stays with a platform admin.""" 216 resp = self._http.request( 217 "POST", 218 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 219 f"/revisions/{revision_id}/review", 220 json={"status": status, "note": note}, 221 ) 222 _raise_for_status(resp) 223 return AdminAuthoredRevisionRow.model_validate(resp.json())
34class AdminAuthoringReviewersResource: 35 """Grant / revoke the right to rule on one competition's uploads.""" 36 37 def __init__(self, http: BaseHttpClient) -> None: 38 self._http = http 39 40 def list( 41 self, competition_id: str, offset: int = 0, limit: int = 50 42 ) -> PagedList[AuthoringReviewerInfo]: 43 """Who may rule on this competition's uploads.""" 44 resp = self._http.request( 45 "GET", 46 f"/admin/competitions/{competition_id}/authoring-reviewers", 47 params={"offset": offset, "limit": limit}, 48 ) 49 _raise_for_status(resp) 50 return _extract_items(resp.json(), AuthoringReviewerInfo) 51 52 def grant( 53 self, competition_id: str, *, user_id: str = "", email: str = "" 54 ) -> AuthoringReviewerInfo: 55 """Grant it, to a user named by ``user_id`` or ``email``. The 56 holder reaches this event's uploads and nothing else of it — not 57 the organiser's console, not the entrants' registration details.""" 58 resp = self._http.request( 59 "POST", 60 f"/admin/competitions/{competition_id}/authoring-reviewers", 61 json={"user_id": user_id, "email": email}, 62 ) 63 _raise_for_status(resp) 64 return AuthoringReviewerInfo.model_validate(resp.json()) 65 66 def revoke(self, competition_id: str, user_id: str) -> None: 67 """Revoke it.""" 68 resp = self._http.request( 69 "DELETE", f"/admin/competitions/{competition_id}/authoring-reviewers/{user_id}" 70 ) 71 _raise_for_status(resp)
Grant / revoke the right to rule on one competition's uploads.
40 def list( 41 self, competition_id: str, offset: int = 0, limit: int = 50 42 ) -> PagedList[AuthoringReviewerInfo]: 43 """Who may rule on this competition's uploads.""" 44 resp = self._http.request( 45 "GET", 46 f"/admin/competitions/{competition_id}/authoring-reviewers", 47 params={"offset": offset, "limit": limit}, 48 ) 49 _raise_for_status(resp) 50 return _extract_items(resp.json(), AuthoringReviewerInfo)
Who may rule on this competition's uploads.
52 def grant( 53 self, competition_id: str, *, user_id: str = "", email: str = "" 54 ) -> AuthoringReviewerInfo: 55 """Grant it, to a user named by ``user_id`` or ``email``. The 56 holder reaches this event's uploads and nothing else of it — not 57 the organiser's console, not the entrants' registration details.""" 58 resp = self._http.request( 59 "POST", 60 f"/admin/competitions/{competition_id}/authoring-reviewers", 61 json={"user_id": user_id, "email": email}, 62 ) 63 _raise_for_status(resp) 64 return AuthoringReviewerInfo.model_validate(resp.json())
Grant it, to a user named by user_id or email. The
holder reaches this event's uploads and nothing else of it — not
the organiser's console, not the entrants' registration details.
66 def revoke(self, competition_id: str, user_id: str) -> None: 67 """Revoke it.""" 68 resp = self._http.request( 69 "DELETE", f"/admin/competitions/{competition_id}/authoring-reviewers/{user_id}" 70 ) 71 _raise_for_status(resp)
Revoke it.
74class AdminCompetitionAuthoredChallengesResource: 75 """Read and rule on one competition's uploads. 76 77 ⚠️ The competition is the **grant**, not a filter: another event's 78 upload answers 404 here, and there is no parameter that widens the 79 listing. 80 """ 81 82 def __init__(self, http: BaseHttpClient) -> None: 83 self._http = http 84 85 def list( 86 self, 87 competition_id: str, 88 *, 89 review_status: ReviewStatus | None = None, 90 smoke: AuthoredSmokeFilter | None = None, 91 revision: ReviewStatus | None = None, 92 order: AuthoredQueueOrder | None = None, 93 destination: AuthoredDestination | None = None, 94 offset: int = 0, 95 limit: int = 50, 96 ) -> PagedList[AdminAuthoredChallengeRow]: 97 """This event's uploads, oldest first. ``smoke``, ``revision``, 98 ``destination`` and ``order`` work as on the platform queue — except that each 99 uploader's ``author_reputation``, and so the ``reputation`` order, 100 counts only the uploads this reviewer may attribute to them.""" 101 params: dict[str, object] = {"offset": offset, "limit": limit} 102 if review_status is not None: 103 params["review_status"] = review_status 104 if smoke is not None: 105 params["smoke"] = smoke 106 if revision is not None: 107 params["revision"] = revision 108 if order is not None: 109 params["order"] = order 110 if destination is not None: 111 params["destination"] = destination 112 resp = self._http.request( 113 "GET", f"/admin/competitions/{competition_id}/authored-challenges", params=params 114 ) 115 _raise_for_status(resp) 116 return _extract_items(resp.json(), AdminAuthoredChallengeRow) 117 118 def get(self, competition_id: str, challenge_id: str) -> AdminAuthoredChallengeDetail: 119 """One upload, with the paths, sizes and execution surface of 120 what is in it.""" 121 resp = self._http.request( 122 "GET", f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 123 ) 124 _raise_for_status(resp) 125 return AdminAuthoredChallengeDetail.model_validate(resp.json()) 126 127 def audit(self, competition_id: str, challenge_id: str) -> AuthoredAuditReport: 128 """Run the challenge audit over this upload. Read ``outcome`` 129 before the counts — ``timed_out`` and ``failed`` are the audit 130 declining to answer, never a clean bill.""" 131 resp = self._http.request( 132 "GET", 133 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/audit", 134 ) 135 _raise_for_status(resp) 136 return AuthoredAuditReport.model_validate(resp.json()) 137 138 def read_file( 139 self, competition_id: str, challenge_id: str, path: str 140 ) -> AuthoredChallengeFileBody: 141 """One file's contents — attacker-authored text written to be 142 run. ``binary`` and ``truncated`` say what you are not seeing.""" 143 resp = self._http.request( 144 "POST", 145 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/file", 146 json={"path": path}, 147 ) 148 _raise_for_status(resp) 149 return AuthoredChallengeFileBody.model_validate(resp.json()) 150 151 def review( 152 self, 153 competition_id: str, 154 challenge_id: str, 155 *, 156 status: ReviewStatus, 157 note: str = "", 158 ) -> AdminAuthoredChallengeRow: 159 """Record a content verdict. ``rejected`` requires a note, and 160 ``pending`` withdraws a verdict. Publishing stays with a 161 platform admin, who reviews what the tree runs.""" 162 resp = self._http.request( 163 "POST", 164 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/review", 165 json={"status": status, "note": note}, 166 ) 167 _raise_for_status(resp) 168 return AdminAuthoredChallengeRow.model_validate(resp.json()) 169 170 def revisions( 171 self, competition_id: str, challenge_id: str 172 ) -> builtins.list[AdminAuthoredRevisionRow]: 173 """Every revision of one of this event's uploads, newest first.""" 174 resp = self._http.request( 175 "GET", 176 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/revisions", 177 ) 178 _raise_for_status(resp) 179 return [AdminAuthoredRevisionRow.model_validate(row) for row in resp.json()] 180 181 def revision_diff( 182 self, competition_id: str, challenge_id: str, revision_id: str 183 ) -> AuthoredRevisionDiff: 184 """A proposed revision against the live tree (plan P3.1).""" 185 resp = self._http.request( 186 "GET", 187 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 188 f"/revisions/{revision_id}/diff", 189 ) 190 _raise_for_status(resp) 191 return AuthoredRevisionDiff.model_validate(resp.json()) 192 193 def read_revision_file( 194 self, competition_id: str, challenge_id: str, revision_id: str, path: str 195 ) -> AuthoredChallengeFileBody: 196 """One file of a revision, whole.""" 197 resp = self._http.request( 198 "POST", 199 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 200 f"/revisions/{revision_id}/file", 201 json={"path": path}, 202 ) 203 _raise_for_status(resp) 204 return AuthoredChallengeFileBody.model_validate(resp.json()) 205 206 def review_revision( 207 self, 208 competition_id: str, 209 challenge_id: str, 210 revision_id: str, 211 *, 212 status: ReviewStatus, 213 note: str = "", 214 ) -> AdminAuthoredRevisionRow: 215 """A content verdict on a revision of one of this event's uploads. 216 Applying it stays with a platform admin.""" 217 resp = self._http.request( 218 "POST", 219 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 220 f"/revisions/{revision_id}/review", 221 json={"status": status, "note": note}, 222 ) 223 _raise_for_status(resp) 224 return AdminAuthoredRevisionRow.model_validate(resp.json())
Read and rule on one competition's uploads.
⚠️ The competition is the grant, not a filter: another event's upload answers 404 here, and there is no parameter that widens the listing.
85 def list( 86 self, 87 competition_id: str, 88 *, 89 review_status: ReviewStatus | None = None, 90 smoke: AuthoredSmokeFilter | None = None, 91 revision: ReviewStatus | None = None, 92 order: AuthoredQueueOrder | None = None, 93 destination: AuthoredDestination | None = None, 94 offset: int = 0, 95 limit: int = 50, 96 ) -> PagedList[AdminAuthoredChallengeRow]: 97 """This event's uploads, oldest first. ``smoke``, ``revision``, 98 ``destination`` and ``order`` work as on the platform queue — except that each 99 uploader's ``author_reputation``, and so the ``reputation`` order, 100 counts only the uploads this reviewer may attribute to them.""" 101 params: dict[str, object] = {"offset": offset, "limit": limit} 102 if review_status is not None: 103 params["review_status"] = review_status 104 if smoke is not None: 105 params["smoke"] = smoke 106 if revision is not None: 107 params["revision"] = revision 108 if order is not None: 109 params["order"] = order 110 if destination is not None: 111 params["destination"] = destination 112 resp = self._http.request( 113 "GET", f"/admin/competitions/{competition_id}/authored-challenges", params=params 114 ) 115 _raise_for_status(resp) 116 return _extract_items(resp.json(), AdminAuthoredChallengeRow)
This event's uploads, oldest first. smoke, revision,
destination and order work as on the platform queue — except that each
uploader's author_reputation, and so the reputation order,
counts only the uploads this reviewer may attribute to them.
118 def get(self, competition_id: str, challenge_id: str) -> AdminAuthoredChallengeDetail: 119 """One upload, with the paths, sizes and execution surface of 120 what is in it.""" 121 resp = self._http.request( 122 "GET", f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 123 ) 124 _raise_for_status(resp) 125 return AdminAuthoredChallengeDetail.model_validate(resp.json())
One upload, with the paths, sizes and execution surface of what is in it.
127 def audit(self, competition_id: str, challenge_id: str) -> AuthoredAuditReport: 128 """Run the challenge audit over this upload. Read ``outcome`` 129 before the counts — ``timed_out`` and ``failed`` are the audit 130 declining to answer, never a clean bill.""" 131 resp = self._http.request( 132 "GET", 133 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/audit", 134 ) 135 _raise_for_status(resp) 136 return AuthoredAuditReport.model_validate(resp.json())
Run the challenge audit over this upload. Read outcome
before the counts — timed_out and failed are the audit
declining to answer, never a clean bill.
138 def read_file( 139 self, competition_id: str, challenge_id: str, path: str 140 ) -> AuthoredChallengeFileBody: 141 """One file's contents — attacker-authored text written to be 142 run. ``binary`` and ``truncated`` say what you are not seeing.""" 143 resp = self._http.request( 144 "POST", 145 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/file", 146 json={"path": path}, 147 ) 148 _raise_for_status(resp) 149 return AuthoredChallengeFileBody.model_validate(resp.json())
One file's contents — attacker-authored text written to be
run. binary and truncated say what you are not seeing.
151 def review( 152 self, 153 competition_id: str, 154 challenge_id: str, 155 *, 156 status: ReviewStatus, 157 note: str = "", 158 ) -> AdminAuthoredChallengeRow: 159 """Record a content verdict. ``rejected`` requires a note, and 160 ``pending`` withdraws a verdict. Publishing stays with a 161 platform admin, who reviews what the tree runs.""" 162 resp = self._http.request( 163 "POST", 164 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/review", 165 json={"status": status, "note": note}, 166 ) 167 _raise_for_status(resp) 168 return AdminAuthoredChallengeRow.model_validate(resp.json())
Record a content verdict. rejected requires a note, and
pending withdraws a verdict. Publishing stays with a
platform admin, who reviews what the tree runs.
170 def revisions( 171 self, competition_id: str, challenge_id: str 172 ) -> builtins.list[AdminAuthoredRevisionRow]: 173 """Every revision of one of this event's uploads, newest first.""" 174 resp = self._http.request( 175 "GET", 176 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}/revisions", 177 ) 178 _raise_for_status(resp) 179 return [AdminAuthoredRevisionRow.model_validate(row) for row in resp.json()]
Every revision of one of this event's uploads, newest first.
181 def revision_diff( 182 self, competition_id: str, challenge_id: str, revision_id: str 183 ) -> AuthoredRevisionDiff: 184 """A proposed revision against the live tree (plan P3.1).""" 185 resp = self._http.request( 186 "GET", 187 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 188 f"/revisions/{revision_id}/diff", 189 ) 190 _raise_for_status(resp) 191 return AuthoredRevisionDiff.model_validate(resp.json())
A proposed revision against the live tree (plan P3.1).
193 def read_revision_file( 194 self, competition_id: str, challenge_id: str, revision_id: str, path: str 195 ) -> AuthoredChallengeFileBody: 196 """One file of a revision, whole.""" 197 resp = self._http.request( 198 "POST", 199 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 200 f"/revisions/{revision_id}/file", 201 json={"path": path}, 202 ) 203 _raise_for_status(resp) 204 return AuthoredChallengeFileBody.model_validate(resp.json())
One file of a revision, whole.
206 def review_revision( 207 self, 208 competition_id: str, 209 challenge_id: str, 210 revision_id: str, 211 *, 212 status: ReviewStatus, 213 note: str = "", 214 ) -> AdminAuthoredRevisionRow: 215 """A content verdict on a revision of one of this event's uploads. 216 Applying it stays with a platform admin.""" 217 resp = self._http.request( 218 "POST", 219 f"/admin/competitions/{competition_id}/authored-challenges/{challenge_id}" 220 f"/revisions/{revision_id}/review", 221 json={"status": status, "note": note}, 222 ) 223 _raise_for_status(resp) 224 return AdminAuthoredRevisionRow.model_validate(resp.json())
A content verdict on a revision of one of this event's uploads. Applying it stays with a platform admin.