diff --git a/.env.example b/.env.example index 079f63c..0d72ac9 100644 --- a/.env.example +++ b/.env.example @@ -23,3 +23,12 @@ CODEX_RESUME_JSON_FLAG_STYLE=before_resume # Optional: set this only if you want plain messages to behave like /ask. ALLOW_PLAIN_TEXT=true + +# Optional Gitea integration for repository, issue, pull request, and git auth workflows. +GITEA_BASE_URL=https://gitea.example.com +GITEA_USERNAME= +GITEA_EMAIL= +GITEA_TOKEN= +GITEA_DEFAULT_OWNER= +GITEA_DEFAULT_REPO= +GITEA_GIT_AUTH_ENABLED=true diff --git a/.gitignore b/.gitignore index 96d0676..c0923aa 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,6 @@ .env +.env.* +!.env.example .venv/ __pycache__/ *.py[cod] diff --git a/Dockerfile b/Dockerfile index a8d43db..b5769ab 100644 --- a/Dockerfile +++ b/Dockerfile @@ -31,6 +31,6 @@ COPY scripts ./scripts COPY README.md ./ RUN mkdir -p /app/data /workspaces /root/.codex \ - && chmod +x /app/scripts/start.sh + && chmod +x /app/scripts/*.sh CMD ["/app/scripts/start.sh"] diff --git a/README.md b/README.md index 3fe4dd1..a43c73a 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ only keeps connection-management commands. /whoami /repo /ask +/work /new /status /settings @@ -34,6 +35,9 @@ only keeps connection-management commands. ``` `ALLOW_PLAIN_TEXT=true` makes regular Telegram messages behave like `/ask`. +Use `/work ` for write-capable tasks such as editing files, committing, +pushing, or opening pull requests. Plain text and `/ask` stay read-oriented by +default. `/settings`, `/models`, `/schedules`, `/records`, and `/sessions` return inline buttons, so you can change speed/model/reasoning settings, delete schedules, @@ -93,6 +97,32 @@ The bridge scheduler runs inside the Telegram bot process. When a schedule is due, it calls Codex app-server and sends the result back to the stored Telegram chat id. +## Gitea integration + +Set these optional values in `.env` to let Codex manage Gitea repositories, +issues, pull requests, and HTTPS git authentication from Telegram: + +```env +GITEA_BASE_URL=https://gitea.example.com +GITEA_USERNAME=your-gitea-user +GITEA_EMAIL=you@example.com +GITEA_TOKEN=your-personal-access-token +GITEA_DEFAULT_OWNER=your-gitea-user-or-org +GITEA_DEFAULT_REPO=your-default-repo +GITEA_GIT_AUTH_ENABLED=true +``` + +Codex receives Gitea MCP tools for repository, issue, branch, and pull-request +metadata. The container also configures `GIT_ASKPASS` so HTTPS `git clone`, +`fetch`, and `push` can use the token without embedding it in command lines. + +Example Telegram workflow: + +```text +/repo nas_connection +/work 이슈 #3을 확인하고 코드 수정 후 테스트해. 변경사항을 codex/gitea-tools 브랜치로 커밋하고 push한 뒤 PR을 열어줘. +``` + ## Deployment 1. Copy `.env.example` to `.env`. diff --git a/app/bot.py b/app/bot.py index a9d778e..b7f09ad 100644 --- a/app/bot.py +++ b/app/bot.py @@ -31,6 +31,7 @@ HELP_TEXT = """Telegram-Codex 연결 브리지 /whoami - 내 Telegram user_id/chat_id 확인 /repo <경로> - /workspaces 아래 작업 폴더 설정 /ask <내용> - 내용을 Codex로 전달 +/work <내용> - 파일 수정, 커밋, 푸시 같은 쓰기 작업 실행 /new - 이 채팅의 Codex 세션 초기화 /status - 브리지 상태 확인 /settings - 모델, 추론강도, 속도 설정 확인 @@ -114,6 +115,8 @@ class TelegramCodexBot: app.add_handler(CommandHandler("whoami", self.whoami)) app.add_handler(CommandHandler("repo", self.repo)) app.add_handler(CommandHandler("ask", self.ask)) + app.add_handler(CommandHandler("work", self.work)) + app.add_handler(CommandHandler("write", self.work)) app.add_handler(CommandHandler("new", self.new)) app.add_handler(CommandHandler("status", self.status)) app.add_handler(CommandHandler("settings", self.settings)) @@ -233,6 +236,19 @@ class TelegramCodexBot: return await self._run_codex_from_update(update, prompt) + async def work(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + if not await self._guard(update): + return + prompt = _args_text(context) + if not prompt: + await self._reply(update, "사용법: /work <파일 수정, 커밋, 푸시 등 작업 내용>") + return + await self._run_codex_from_update( + update, + prompt, + sandbox=self.config.write_sandbox, + ) + async def new(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await self._guard(update): return @@ -877,14 +893,19 @@ class TelegramCodexBot: return await self._run_codex_from_update(update, text) - async def _run_codex_from_update(self, update: Update, prompt: str) -> None: + async def _run_codex_from_update( + self, + update: Update, + prompt: str, + sandbox: str | None = None, + ) -> None: chat_id = update.effective_chat.id if update.effective_chat: await update.effective_chat.send_action(ChatAction.TYPING) result, fallback_note = await self._run_codex_for_chat( chat_id=chat_id, prompt=prompt, - sandbox=self.config.readonly_sandbox, + sandbox=sandbox or self.config.readonly_sandbox, update_chat_state=True, ) await self._reply_result(update, result, fallback_note) @@ -929,13 +950,14 @@ class TelegramCodexBot: chat_id=chat_id, chat_state=chat_state, cwd=cwd, + sandbox=sandbox, ) temp_state = ChatState( repo_path=chat_state.repo_path, session_id=session_id_override or chat_state.session_id, session_backend="app-server" if session_id_override else chat_state.session_backend, ) - skip_git_repo_check = chat_state.repo_path is None and sandbox == self.config.readonly_sandbox + skip_git_repo_check = chat_state.repo_path is None result, backend_used, fallback_note = await self._run_with_backend( chat_id=chat_id, prompt=bridge_prompt, @@ -1127,6 +1149,7 @@ class TelegramCodexBot: chat_id: int, chat_state: ChatState, cwd: Path, + sandbox: str, ) -> str: thread_id = chat_state.session_id or "(not assigned yet)" schedules = self.state.list_schedules(chat_id=chat_id) @@ -1138,14 +1161,25 @@ class TelegramCodexBot: if not record_lines: record_lines = "- none" runtime_settings = self._format_runtime_settings() + gitea_context = self._format_gitea_context() return ( "[Telegram bridge context]\n" "You are connected to the user through a private Telegram bridge.\n" f"Default Telegram chat_id for replies, schedules, and notifications: {chat_id}\n" f"Current Codex app-server thread_id: {thread_id}\n" f"Current working directory: {cwd}\n" + f"Current sandbox: {sandbox}\n" "Current runtime settings:\n" f"{runtime_settings}\n" + "Configured Gitea integration:\n" + f"{gitea_context}\n" + "If the user asks for repository, issue, branch, or pull-request metadata in Gitea, " + "use the gitea_* MCP tools. Never reveal GITEA_TOKEN or print commands that embed it. " + "For HTTPS git clone/fetch/push, the container may provide GIT_ASKPASS from the " + "configured Gitea environment.\n" + "Only edit files, commit, or push when this turn is running with a write-capable " + "sandbox, typically from the Telegram /work command. In read-only turns, inspect and " + "explain instead of attempting file edits.\n" "If the user asks to schedule, remind, monitor, or notify later, use the " "telegram_create_schedule MCP tool. Use the default chat_id above and do not ask " "for a destination unless the user explicitly wants a different chat. The schedule " @@ -1168,6 +1202,19 @@ class TelegramCodexBot: f"{user_prompt}" ) + def _format_gitea_context(self) -> str: + if not self.config.gitea_base_url: + return "- not configured" + lines = [ + f"- base_url={self.config.gitea_base_url}", + f"- username={self.config.gitea_username or '(not set)'}", + f"- email={self.config.gitea_email or '(not set)'}", + f"- default_owner={self.config.gitea_default_owner or '(not set)'}", + f"- default_repo={self.config.gitea_default_repo or '(not set)'}", + f"- token={'configured' if self.config.gitea_token_configured else 'missing'}", + ] + return "\n".join(lines) + def _should_use_app_server(self, sandbox: str) -> bool: if self.config.codex_backend == "exec": return False diff --git a/app/bridge_mcp.py b/app/bridge_mcp.py index cb98bf3..9fed78f 100644 --- a/app/bridge_mcp.py +++ b/app/bridge_mcp.py @@ -6,6 +6,7 @@ import sys from pathlib import Path from typing import Any +from app.gitea_client import GiteaClient from app.state import StateStore, format_personal_record, format_schedule @@ -58,7 +59,9 @@ class BridgeMcpServer: "Use these tools when the user asks to schedule recurring Telegram " "notifications, manage bridge schedules, or persist personal assistant " "records such as work history, todos, purchase candidates, birthdays, " - "contacts, and notes. The Telegram chat_id and " + "contacts, and notes. Use the Gitea tools when the user asks to inspect " + "or manage repositories, issues, pull requests, and code-review workflow " + "metadata in the user's configured Gitea instance. The Telegram chat_id and " "current thread_id are provided in the bridge context inside each user turn. " "Use that chat_id as the default notification target and do not ask for a " "destination unless the user explicitly wants a different chat." @@ -106,6 +109,22 @@ class BridgeMcpServer: return self.update_personal_record(arguments) if name == "telegram_delete_personal_record": return self.delete_personal_record(arguments) + if name == "gitea_list_repositories": + return self.gitea_list_repositories(arguments) + if name == "gitea_get_repository": + return self.gitea_get_repository(arguments) + if name == "gitea_list_issues": + return self.gitea_list_issues(arguments) + if name == "gitea_create_issue": + return self.gitea_create_issue(arguments) + if name == "gitea_comment_issue": + return self.gitea_comment_issue(arguments) + if name == "gitea_list_pull_requests": + return self.gitea_list_pull_requests(arguments) + if name == "gitea_create_pull_request": + return self.gitea_create_pull_request(arguments) + if name == "gitea_list_branches": + return self.gitea_list_branches(arguments) raise ValueError(f"Unknown tool: {name}") def create_schedule(self, arguments: dict[str, Any]) -> dict[str, Any]: @@ -251,6 +270,132 @@ class BridgeMcpServer: "structuredContent": {"deleted": removed}, } + def gitea_list_repositories(self, arguments: dict[str, Any]) -> dict[str, Any]: + client = GiteaClient.from_env() + repos = client.list_repositories( + limit=int(arguments.get("limit") or 30), + page=int(arguments.get("page") or 1), + ) + if not repos: + text = "No Gitea repositories found." + else: + text = "\n".join(_format_gitea_repo(repo) for repo in repos) + return { + "content": [{"type": "text", "text": text}], + "structuredContent": {"repositories": repos}, + } + + def gitea_get_repository(self, arguments: dict[str, Any]) -> dict[str, Any]: + client = GiteaClient.from_env() + owner, repo = _repo_arguments(arguments, client) + repository = client.get_repository(owner, repo) + return { + "content": [{"type": "text", "text": _format_gitea_repo(repository)}], + "structuredContent": {"repository": repository}, + } + + def gitea_list_issues(self, arguments: dict[str, Any]) -> dict[str, Any]: + client = GiteaClient.from_env() + owner, repo = _repo_arguments(arguments, client) + issues = client.list_issues( + owner, + repo, + state=str(arguments.get("state") or "open"), + limit=int(arguments.get("limit") or 20), + page=int(arguments.get("page") or 1), + ) + if not issues: + text = "No Gitea issues found." + else: + text = "\n".join(_format_gitea_issue(issue) for issue in issues) + return { + "content": [{"type": "text", "text": text}], + "structuredContent": {"issues": issues}, + } + + def gitea_create_issue(self, arguments: dict[str, Any]) -> dict[str, Any]: + client = GiteaClient.from_env() + owner, repo = _repo_arguments(arguments, client) + issue = client.create_issue( + owner, + repo, + title=str(arguments["title"]), + body=str(arguments.get("body") or ""), + ) + return { + "content": [{"type": "text", "text": f"Created Gitea issue: {_format_gitea_issue(issue)}"}], + "structuredContent": {"issue": issue}, + } + + def gitea_comment_issue(self, arguments: dict[str, Any]) -> dict[str, Any]: + client = GiteaClient.from_env() + owner, repo = _repo_arguments(arguments, client) + comment = client.comment_issue( + owner, + repo, + issue_index=int(arguments["issue_index"]), + body=str(arguments["body"]), + ) + url = comment.get("html_url") or comment.get("url") or "" + text = f"Added Gitea issue comment: {url}" if url else "Added Gitea issue comment." + return { + "content": [{"type": "text", "text": text}], + "structuredContent": {"comment": comment}, + } + + def gitea_list_pull_requests(self, arguments: dict[str, Any]) -> dict[str, Any]: + client = GiteaClient.from_env() + owner, repo = _repo_arguments(arguments, client) + pulls = client.list_pull_requests( + owner, + repo, + state=str(arguments.get("state") or "open"), + limit=int(arguments.get("limit") or 20), + page=int(arguments.get("page") or 1), + ) + if not pulls: + text = "No Gitea pull requests found." + else: + text = "\n".join(_format_gitea_pull_request(pull) for pull in pulls) + return { + "content": [{"type": "text", "text": text}], + "structuredContent": {"pull_requests": pulls}, + } + + def gitea_create_pull_request(self, arguments: dict[str, Any]) -> dict[str, Any]: + client = GiteaClient.from_env() + owner, repo = _repo_arguments(arguments, client) + pull = client.create_pull_request( + owner, + repo, + head=str(arguments["head"]), + base=str(arguments.get("base") or "main"), + title=str(arguments["title"]), + body=str(arguments.get("body") or ""), + ) + return { + "content": [{"type": "text", "text": f"Created Gitea pull request: {_format_gitea_pull_request(pull)}"}], + "structuredContent": {"pull_request": pull}, + } + + def gitea_list_branches(self, arguments: dict[str, Any]) -> dict[str, Any]: + client = GiteaClient.from_env() + owner, repo = _repo_arguments(arguments, client) + branches = client.list_branches( + owner, + repo, + limit=int(arguments.get("limit") or 30), + page=int(arguments.get("page") or 1), + ) + if not branches: + text = "No Gitea branches found." + else: + text = "\n".join(str(branch.get("name") or "(unknown)") for branch in branches) + return { + "content": [{"type": "text", "text": text}], + "structuredContent": {"branches": branches}, + } + def _tools() -> list[dict[str, Any]]: return [ @@ -453,6 +598,145 @@ def _tools() -> list[dict[str, Any]]: "openWorldHint": False, }, }, + ] + _gitea_tools() + + +def _gitea_tools() -> list[dict[str, Any]]: + repo_properties = { + "owner": { + "type": "string", + "description": "Repository owner. Defaults to GITEA_DEFAULT_OWNER when omitted.", + }, + "repo": { + "type": "string", + "description": "Repository name. Defaults to GITEA_DEFAULT_REPO when omitted.", + }, + } + readonly_annotations = { + "readOnlyHint": True, + "destructiveHint": False, + "idempotentHint": True, + "openWorldHint": False, + } + write_annotations = { + "readOnlyHint": False, + "destructiveHint": False, + "idempotentHint": False, + "openWorldHint": False, + } + return [ + { + "name": "gitea_list_repositories", + "description": "List repositories visible to the configured Gitea token.", + "inputSchema": { + "type": "object", + "properties": { + "limit": {"type": "integer", "minimum": 1, "maximum": 100}, + "page": {"type": "integer", "minimum": 1}, + }, + }, + "annotations": {"title": "List Gitea repositories", **readonly_annotations}, + }, + { + "name": "gitea_get_repository", + "description": "Get metadata for one Gitea repository.", + "inputSchema": { + "type": "object", + "properties": repo_properties, + }, + "annotations": {"title": "Get Gitea repository", **readonly_annotations}, + }, + { + "name": "gitea_list_issues", + "description": "List issues for a Gitea repository.", + "inputSchema": { + "type": "object", + "properties": { + **repo_properties, + "state": {"type": "string", "enum": ["open", "closed", "all"]}, + "limit": {"type": "integer", "minimum": 1, "maximum": 100}, + "page": {"type": "integer", "minimum": 1}, + }, + }, + "annotations": {"title": "List Gitea issues", **readonly_annotations}, + }, + { + "name": "gitea_create_issue", + "description": "Create an issue in a Gitea repository.", + "inputSchema": { + "type": "object", + "properties": { + **repo_properties, + "title": {"type": "string"}, + "body": {"type": "string"}, + }, + "required": ["title"], + }, + "annotations": {"title": "Create Gitea issue", **write_annotations}, + }, + { + "name": "gitea_comment_issue", + "description": "Add a comment to a Gitea issue or pull request conversation.", + "inputSchema": { + "type": "object", + "properties": { + **repo_properties, + "issue_index": {"type": "integer", "minimum": 1}, + "body": {"type": "string"}, + }, + "required": ["issue_index", "body"], + }, + "annotations": {"title": "Comment on Gitea issue", **write_annotations}, + }, + { + "name": "gitea_list_pull_requests", + "description": "List pull requests for a Gitea repository.", + "inputSchema": { + "type": "object", + "properties": { + **repo_properties, + "state": {"type": "string", "enum": ["open", "closed", "all"]}, + "limit": {"type": "integer", "minimum": 1, "maximum": 100}, + "page": {"type": "integer", "minimum": 1}, + }, + }, + "annotations": {"title": "List Gitea pull requests", **readonly_annotations}, + }, + { + "name": "gitea_create_pull_request", + "description": "Create a pull request in a Gitea repository after a branch has been pushed.", + "inputSchema": { + "type": "object", + "properties": { + **repo_properties, + "head": { + "type": "string", + "description": "Source branch, for example codex/my-change.", + }, + "base": { + "type": "string", + "description": "Target branch. Defaults to main.", + }, + "title": {"type": "string"}, + "body": {"type": "string"}, + }, + "required": ["head", "title"], + }, + "annotations": {"title": "Create Gitea pull request", **write_annotations}, + }, + { + "name": "gitea_list_branches", + "description": "List branches for a Gitea repository.", + "inputSchema": { + "type": "object", + "properties": { + **repo_properties, + "limit": {"type": "integer", "minimum": 1, "maximum": 100}, + "page": {"type": "integer", "minimum": 1}, + }, + }, + "annotations": {"title": "List Gitea branches", **readonly_annotations}, + }, ] @@ -460,6 +744,49 @@ def _record_categories() -> list[str]: return ["work_log", "todo", "purchase", "birthday", "contact", "note", "preference"] +def _repo_arguments(arguments: dict[str, Any], client: GiteaClient) -> tuple[str, str]: + owner = str(arguments.get("owner") or client.config.default_owner or "").strip() + repo = str(arguments.get("repo") or client.config.default_repo or "").strip() + if not owner: + raise ValueError("owner is required or GITEA_DEFAULT_OWNER must be set") + if not repo: + raise ValueError("repo is required or GITEA_DEFAULT_REPO must be set") + return owner, repo + + +def _format_gitea_repo(repo: dict[str, Any]) -> str: + full_name = repo.get("full_name") or repo.get("name") or "(unknown)" + default_branch = repo.get("default_branch") or repo.get("default_branch_name") or "" + clone_url = repo.get("clone_url") or "" + html_url = repo.get("html_url") or "" + bits = [str(full_name)] + if default_branch: + bits.append(f"default={default_branch}") + if html_url: + bits.append(str(html_url)) + if clone_url: + bits.append(f"clone={clone_url}") + return " - ".join(bits) + + +def _format_gitea_issue(issue: dict[str, Any]) -> str: + index = issue.get("number") or issue.get("index") or "?" + title = issue.get("title") or "(untitled)" + state = issue.get("state") or "unknown" + html_url = issue.get("html_url") or "" + suffix = f" - {html_url}" if html_url else "" + return f"#{index} [{state}] {title}{suffix}" + + +def _format_gitea_pull_request(pull: dict[str, Any]) -> str: + index = pull.get("number") or pull.get("index") or "?" + title = pull.get("title") or "(untitled)" + state = pull.get("state") or "unknown" + html_url = pull.get("html_url") or "" + suffix = f" - {html_url}" if html_url else "" + return f"#{index} [{state}] {title}{suffix}" + + def _result(request_id: Any, result: dict[str, Any]) -> dict[str, Any]: return {"jsonrpc": "2.0", "id": request_id, "result": result} diff --git a/app/config.py b/app/config.py index d49845e..dd8fae6 100644 --- a/app/config.py +++ b/app/config.py @@ -52,6 +52,12 @@ class Config: timezone: str schedule_poll_seconds: int persistent_ws_idle_seconds: int + gitea_base_url: str | None + gitea_username: str | None + gitea_email: str | None + gitea_default_owner: str | None + gitea_default_repo: str | None + gitea_token_configured: bool @classmethod def load(cls) -> "Config": @@ -93,4 +99,10 @@ class Config: timezone=os.getenv("BOT_TIMEZONE", "Asia/Seoul"), schedule_poll_seconds=_int_env("SCHEDULE_POLL_SECONDS", 30), persistent_ws_idle_seconds=_int_env("PERSISTENT_WS_IDLE_SECONDS", 600), + gitea_base_url=os.getenv("GITEA_BASE_URL", "").strip() or None, + gitea_username=os.getenv("GITEA_USERNAME", "").strip() or None, + gitea_email=os.getenv("GITEA_EMAIL", "").strip() or None, + gitea_default_owner=os.getenv("GITEA_DEFAULT_OWNER", "").strip() or None, + gitea_default_repo=os.getenv("GITEA_DEFAULT_REPO", "").strip() or None, + gitea_token_configured=bool(os.getenv("GITEA_TOKEN", "").strip()), ) diff --git a/app/gitea_client.py b/app/gitea_client.py new file mode 100644 index 0000000..64ae0a6 --- /dev/null +++ b/app/gitea_client.py @@ -0,0 +1,198 @@ +from __future__ import annotations + +import json +import os +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass +from typing import Any + + +class GiteaError(RuntimeError): + pass + + +@dataclass(frozen=True) +class GiteaConfig: + base_url: str + token: str + username: str | None = None + email: str | None = None + default_owner: str | None = None + default_repo: str | None = None + + @classmethod + def from_env(cls) -> "GiteaConfig": + base_url = os.getenv("GITEA_BASE_URL", "").strip().rstrip("/") + token = os.getenv("GITEA_TOKEN", "").strip() + if not base_url: + raise GiteaError("GITEA_BASE_URL is required") + if not token: + raise GiteaError("GITEA_TOKEN is required") + return cls( + base_url=base_url, + token=token, + username=os.getenv("GITEA_USERNAME", "").strip() or None, + email=os.getenv("GITEA_EMAIL", "").strip() or None, + default_owner=os.getenv("GITEA_DEFAULT_OWNER", "").strip() or None, + default_repo=os.getenv("GITEA_DEFAULT_REPO", "").strip() or None, + ) + + +class GiteaClient: + def __init__(self, config: GiteaConfig) -> None: + self.config = config + + @classmethod + def from_env(cls) -> "GiteaClient": + return cls(GiteaConfig.from_env()) + + def list_repositories(self, limit: int = 30, page: int = 1) -> list[dict[str, Any]]: + result = self._request( + "GET", + "/user/repos", + query={"limit": str(limit), "page": str(page)}, + ) + return result if isinstance(result, list) else [] + + def get_repository(self, owner: str, repo: str) -> dict[str, Any]: + result = self._request("GET", f"/repos/{_quote(owner)}/{_quote(repo)}") + return result if isinstance(result, dict) else {} + + def list_issues( + self, + owner: str, + repo: str, + state: str = "open", + limit: int = 20, + page: int = 1, + ) -> list[dict[str, Any]]: + result = self._request( + "GET", + f"/repos/{_quote(owner)}/{_quote(repo)}/issues", + query={"state": state, "limit": str(limit), "page": str(page)}, + ) + return result if isinstance(result, list) else [] + + def create_issue( + self, + owner: str, + repo: str, + title: str, + body: str = "", + ) -> dict[str, Any]: + result = self._request( + "POST", + f"/repos/{_quote(owner)}/{_quote(repo)}/issues", + body={"title": title, "body": body}, + ) + return result if isinstance(result, dict) else {} + + def comment_issue( + self, + owner: str, + repo: str, + issue_index: int, + body: str, + ) -> dict[str, Any]: + result = self._request( + "POST", + f"/repos/{_quote(owner)}/{_quote(repo)}/issues/{issue_index}/comments", + body={"body": body}, + ) + return result if isinstance(result, dict) else {} + + def list_pull_requests( + self, + owner: str, + repo: str, + state: str = "open", + limit: int = 20, + page: int = 1, + ) -> list[dict[str, Any]]: + result = self._request( + "GET", + f"/repos/{_quote(owner)}/{_quote(repo)}/pulls", + query={"state": state, "limit": str(limit), "page": str(page)}, + ) + return result if isinstance(result, list) else [] + + def create_pull_request( + self, + owner: str, + repo: str, + head: str, + base: str, + title: str, + body: str = "", + ) -> dict[str, Any]: + result = self._request( + "POST", + f"/repos/{_quote(owner)}/{_quote(repo)}/pulls", + body={"head": head, "base": base, "title": title, "body": body}, + ) + return result if isinstance(result, dict) else {} + + def list_branches(self, owner: str, repo: str, limit: int = 30, page: int = 1) -> list[dict[str, Any]]: + result = self._request( + "GET", + f"/repos/{_quote(owner)}/{_quote(repo)}/branches", + query={"limit": str(limit), "page": str(page)}, + ) + return result if isinstance(result, list) else [] + + def _request( + self, + method: str, + path: str, + body: dict[str, Any] | None = None, + query: dict[str, str] | None = None, + ) -> Any: + url = self._url(path, query) + data = None + headers = { + "Accept": "application/json", + "Authorization": f"token {self.config.token}", + "User-Agent": "telegram-codex-bridge", + } + if body is not None: + data = json.dumps(body).encode("utf-8") + headers["Content-Type"] = "application/json" + + request = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(request, timeout=30) as response: + raw = response.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace") + raise GiteaError(f"Gitea API failed ({exc.code}): {_short_error(detail)}") from exc + except urllib.error.URLError as exc: + raise GiteaError(f"Gitea API connection failed: {exc.reason}") from exc + + if not raw: + return None + try: + return json.loads(raw) + except json.JSONDecodeError as exc: + raise GiteaError(f"Gitea API returned invalid JSON: {exc}") from exc + + def _url(self, path: str, query: dict[str, str] | None = None) -> str: + base = self.config.base_url + if not base.endswith("/api/v1"): + base = base + "/api/v1" + url = base + path + if query: + url += "?" + urllib.parse.urlencode(query) + return url + + +def _quote(value: str) -> str: + return urllib.parse.quote(value, safe="") + + +def _short_error(text: str, limit: int = 600) -> str: + text = text.strip() + if len(text) <= limit: + return text + return text[:limit] + "..." diff --git a/scripts/gitea-askpass.sh b/scripts/gitea-askpass.sh new file mode 100644 index 0000000..046ee90 --- /dev/null +++ b/scripts/gitea-askpass.sh @@ -0,0 +1,13 @@ +#!/bin/sh +set -eu + +prompt="${1:-}" + +case "$prompt" in + *Username*|*username*) + printf '%s\n' "${GITEA_USERNAME:-git}" + ;; + *) + printf '%s\n' "${GITEA_TOKEN:-}" + ;; +esac diff --git a/scripts/start.sh b/scripts/start.sh index c5779d3..a53bc26 100644 --- a/scripts/start.sh +++ b/scripts/start.sh @@ -9,6 +9,7 @@ CODEX_APP_SERVER_ENABLED="${CODEX_APP_SERVER_ENABLED:-true}" CODEX_APP_SERVER_CWD="${CODEX_APP_SERVER_CWD:-$BOT_WORKSPACE_ROOT}" CODEX_APP_SERVER_SANDBOX="${CODEX_APP_SERVER_SANDBOX:-${CODEX_READONLY_SANDBOX:-read-only}}" CODEX_APP_SERVER_APPROVAL_POLICY="${CODEX_APP_SERVER_APPROVAL_POLICY:-never}" +GITEA_GIT_AUTH_ENABLED="${GITEA_GIT_AUTH_ENABLED:-true}" APP_SERVER_PID="" @@ -22,6 +23,18 @@ cleanup() { trap cleanup INT TERM EXIT mkdir -p /app/data "$BOT_WORKSPACE_ROOT" "${CODEX_HOME:-/root/.codex}" + +if [ "$GITEA_GIT_AUTH_ENABLED" = "true" ] && [ -n "${GITEA_TOKEN:-}" ]; then + export GIT_ASKPASS="/app/scripts/gitea-askpass.sh" + export GIT_TERMINAL_PROMPT=0 + if [ -n "${GITEA_USERNAME:-}" ]; then + git config --global user.name "$GITEA_USERNAME" || true + fi + if [ -n "${GITEA_EMAIL:-}" ]; then + git config --global user.email "$GITEA_EMAIL" || true + fi +fi + python -m app.ensure_codex_config if [ "$CODEX_APP_SERVER_ENABLED" = "true" ]; then