| """Strictly scoped GitHub sandbox text creation; no model-generated action executes here.""" |
| from __future__ import annotations |
| import base64 |
| import hashlib |
| import json |
| import os |
| import re |
| import secrets |
| import threading |
| import time |
| import urllib.error |
| import urllib.parse |
| import urllib.request |
|
|
| OWNER = "kaministrator999-ui" |
| REPO = "Swrlzing-core" |
| BRANCH = "main" |
| ROOT = "write-action-tests/" |
| MAX_BYTES = 4096 |
| _pending = {} |
| _lock = threading.RLock() |
| _ttl = 300 |
|
|
|
|
| class WriteError(Exception): |
| def __init__(self, code, status=400): |
| super().__init__(code) |
| self.code, self.status = code, status |
|
|
|
|
| def _authorized(user): |
| |
| |
| from github_connection import _sandbox_authorize_user |
| from fastapi import HTTPException |
| try: |
| return _sandbox_authorize_user(user) |
| except HTTPException as exc: |
| raise WriteError(str(exc.detail), exc.status_code) from None |
|
|
|
|
| def _path(filename): |
| if not isinstance(filename, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}\.txt", filename, re.ASCII): |
| raise WriteError("INVALID_TXT_FILENAME") |
| if filename.startswith(".") or ".." in filename: |
| raise WriteError("INVALID_TXT_FILENAME") |
| return ROOT + filename |
|
|
|
|
| def _content(content): |
| if not isinstance(content, str) or not content or len(content.encode("utf-8")) > MAX_BYTES: |
| raise WriteError("INVALID_TEXT_CONTENT") |
| return content |
|
|
|
|
| def _request(method, path, token, data=None): |
| url = "https://api.github.com/repos/" + OWNER + "/" + REPO + "/contents/" + urllib.parse.quote(path, safe="/") + "?ref=" + BRANCH if method == "GET" else "https://api.github.com/repos/" + OWNER + "/" + REPO + "/contents/" + urllib.parse.quote(path, safe="/") |
| headers = {"Authorization": "Bearer " + token, "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", "User-Agent": "swrlz-sandbox-create-txt-v1"} |
| if data is not None: |
| headers["Content-Type"] = "application/json" |
| req = urllib.request.Request(url, method=method, headers=headers, data=json.dumps(data).encode("utf-8") if data is not None else None) |
| try: |
| with urllib.request.urlopen(req, timeout=15) as response: |
| return response.status, json.loads(response.read().decode("utf-8")) |
| except urllib.error.HTTPError as error: |
| if error.code == 404: |
| return 404, None |
| if error.code in (409, 422): |
| raise WriteError("GITHUB_CONFLICT_OR_EXISTS", 409) from None |
| raise WriteError("GITHUB_WRITE_UNAVAILABLE", 502) from None |
| except (urllib.error.URLError, TimeoutError, ValueError): |
| raise WriteError("GITHUB_WRITE_UNAVAILABLE", 502) from None |
|
|
|
|
| def prepare(user, filename, content): |
| token = _authorized(user) |
| path = _path(filename) |
| content = _content(content) |
| status, _ = _request("GET", path, token) |
| if status != 404: |
| raise WriteError("FILE_ALREADY_EXISTS", 409) |
| ticket = secrets.token_urlsafe(32) |
| digest = hashlib.sha256(content.encode("utf-8")).hexdigest() |
| now = time.monotonic() |
| with _lock: |
| for key, item in list(_pending.items()): |
| if item["expires"] < now: |
| del _pending[key] |
| _pending[ticket] = {"userId": user["id"], "path": path, "content": content, "digest": digest, "expires": now + _ttl} |
| return {"approvalToken": ticket, "expiresInSeconds": _ttl, "operation": "CREATE_FILE", "repo": OWNER + "/" + REPO, "branch": BRANCH, "path": path, "preview": content, "sha256": digest} |
|
|
|
|
| def confirm(user, ticket, approved): |
| token = _authorized(user) |
| if not isinstance(ticket, str) or not ticket: |
| raise WriteError("INVALID_APPROVAL_TOKEN") |
| with _lock: |
| item = _pending.pop(ticket, None) |
| if not item or item["expires"] < time.monotonic() or item["userId"] != user["id"]: |
| raise WriteError("APPROVAL_EXPIRED_OR_INVALID", 403) |
| if approved is not True: |
| return {"ok": False, "code": "USER_DECLINED"} |
| status, _ = _request("GET", item["path"], token) |
| if status != 404: |
| raise WriteError("FILE_ALREADY_EXISTS", 409) |
| |
| content_bytes = item["content"].encode("utf-8") |
| status, response = _request("PUT", item["path"], token, {"message": "test(sandbox): approved CREATE_FILE " + item["path"].split("/")[-1], "branch": BRANCH, "content": base64.b64encode(content_bytes).decode("ascii")}) |
| if status not in (200, 201): |
| raise WriteError("GITHUB_WRITE_UNAVAILABLE", 502) |
| commit = str((response or {}).get("commit", {}).get("sha") or "") |
| status, fetched = _request("GET", item["path"], token) |
| if status != 200 or not isinstance(fetched, dict): |
| raise WriteError("GITHUB_READBACK_FAILED", 502) |
| try: |
| observed = base64.b64decode(fetched["content"].replace("\n", ""), validate=False) |
| except (KeyError, TypeError, ValueError): |
| raise WriteError("GITHUB_READBACK_FAILED", 502) from None |
| verified = hashlib.sha256(observed).hexdigest() == item["digest"] and observed == content_bytes |
| if not verified: |
| raise WriteError("GITHUB_READBACK_MISMATCH", 502) |
| receipt = {"ok": True, "operation": "CREATE_FILE", "repo": OWNER + "/" + REPO, "branch": BRANCH, "path": item["path"], "commitSha": commit, "verified": True, "url": "https://github.com/" + OWNER + "/" + REPO + "/blob/" + BRANCH + "/" + item["path"]} |
| |
| |
| from github_connection import _api |
| from github_work_summary_v260 import make_work_widget |
| receipt["widget"] = make_work_widget(receipt, item["content"], token, github_get=_api) |
| return receipt |
|
|
|
|
|
|
| def create_requested_test_file(user, request_id, prompt): |
| """Exactly one CREATE_FILE for a direct, owner-authorized Chat test request. |
| |
| No generated/model/retrieved text enters this path. The filename is stable |
| for the accepted request ID to avoid duplicate files on retries. This is |
| the narrow exception to the separate preview/confirm panel; unrelated |
| writes still require that panel, and the GitHub scope check is unchanged. |
| """ |
| from github_txt_intent_v256 import wants_direct_github_test_txt |
| if not wants_direct_github_test_txt(prompt): |
| raise WriteError("DIRECT_TEST_FILE_REQUEST_REQUIRED", 403) |
| if not isinstance(user, dict) or not isinstance(user.get("id"), str) or not user["id"].startswith("google:"): |
| raise WriteError("SANDBOX_WRITE_OWNER_REQUIRED", 403) |
| if not isinstance(request_id, str) or not 0 < len(request_id) <= 160: |
| raise WriteError("INVALID_REQUEST_ID", 400) |
| |
| |
| suffix = hashlib.sha256((user["id"] + "\0" + request_id).encode("utf-8")).hexdigest()[:16] |
| filename = "chat-test-" + suffix + ".txt" |
| contents = ( |
| "Hello from §wyrlz Chat!\n" |
| "This is a harmless GitHub text-file creation test, requested directly " |
| "in Chat by the connected repository owner.\n" |
| "Target: Swrlzing-core / main / write-action-tests/\n" |
| ) |
| |
| |
| preview = prepare(user, filename, contents) |
| return confirm(user, preview["approvalToken"], True) |
|
|
