refactor: rename auth.py to gitea_auth.py and ignore env files
This commit is contained in:
@@ -1,3 +1,6 @@
|
||||
venv/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
.env*
|
||||
.vscode/
|
||||
graphify-out/
|
||||
|
||||
@@ -11,12 +11,24 @@ A collection of Python scripts and an MCP server to automate interactions with G
|
||||
|
||||
## Authentication
|
||||
|
||||
Scripts extract credentials from the macOS keychain automatically — no tokens on the command line.
|
||||
Authentication is configured via environment variables or a local `.env` file in the repository root (uses `python-dotenv`).
|
||||
|
||||
- **dadeschools** — HTTPS via `git credential fill` (SSH:2222 is flaky)
|
||||
- **prgs** — SSH via `ssh://git@gitea-ssh.prgs.cc:2222` (SSH is reliable here)
|
||||
Create a `.env` file in the project root:
|
||||
|
||||
Ensure you've logged in via Git over HTTPS at least once so the keychain caches your credentials.
|
||||
```bash
|
||||
# Option A: Gitea Personal Access Tokens (Recommended)
|
||||
GITEA_TOKEN_DADESCHOOLS="your_token_here"
|
||||
GITEA_TOKEN_PRGS="your_token_here"
|
||||
|
||||
# Option B: Gitea Username & Password (fallback)
|
||||
GITEA_USER_DADESCHOOLS="username"
|
||||
GITEA_PASS_DADESCHOOLS="password"
|
||||
GITEA_USER_PRGS="username"
|
||||
GITEA_PASS_PRGS="password"
|
||||
|
||||
# Optional: Fallback to macOS Keychain (via git credential fill)
|
||||
# GITEA_USE_KEYCHAIN=1
|
||||
```
|
||||
|
||||
## MCP Server (Recommended)
|
||||
|
||||
@@ -158,7 +170,7 @@ Use `--help` on any Python script or shell script for full usage details.
|
||||
## Architecture
|
||||
|
||||
```
|
||||
auth.py ← shared auth & API helpers (get_credentials, api_request)
|
||||
gitea_auth.py ← shared auth & API helpers (get_credentials, api_request)
|
||||
mcp_server.py ← MCP server (FastMCP, stdio transport)
|
||||
create_issue.py ← CLI: create issues
|
||||
create_pr.py ← CLI: create PRs
|
||||
|
||||
@@ -1,93 +0,0 @@
|
||||
"""Shared authentication and API helper for Gitea scripts.
|
||||
|
||||
Pulls credentials from the macOS keychain via `git credential fill`
|
||||
so no tokens appear on the command line.
|
||||
"""
|
||||
import json
|
||||
import base64
|
||||
import subprocess
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
|
||||
# Known Gitea instances — shared by all scripts.
|
||||
REMOTES = {
|
||||
"dadeschools": {
|
||||
"host": "gitea.dadeschools.net",
|
||||
"org": "Contractor",
|
||||
"repo": "Timesheet",
|
||||
},
|
||||
"prgs": {
|
||||
"host": "gitea.prgs.cc",
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Timesheet",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def get_credentials(host):
|
||||
"""Return (user, password) for *host* via ``git credential fill``."""
|
||||
p = subprocess.Popen(
|
||||
["git", "credential", "fill"],
|
||||
stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True,
|
||||
)
|
||||
out, _ = p.communicate(f"protocol=https\nhost={host}\n\n")
|
||||
user = password = ""
|
||||
for line in out.splitlines():
|
||||
if line.startswith("username="):
|
||||
user = line.split("=", 1)[1]
|
||||
elif line.startswith("password="):
|
||||
password = line.split("=", 1)[1]
|
||||
return user, password
|
||||
|
||||
|
||||
def get_auth_header(host):
|
||||
"""Return an ``Authorization: Basic …`` header value for *host*."""
|
||||
user, password = get_credentials(host)
|
||||
if not user or not password:
|
||||
return None
|
||||
token = base64.b64encode(f"{user}:{password}".encode()).decode()
|
||||
return f"Basic {token}"
|
||||
|
||||
|
||||
def resolve_remote(args):
|
||||
"""Given parsed argparse args with --remote/--host/--org/--repo,
|
||||
return (host, org, repo) with overrides applied."""
|
||||
profile = REMOTES[args.remote]
|
||||
host = args.host or profile["host"]
|
||||
org = args.org or profile["org"]
|
||||
repo = args.repo or profile["repo"]
|
||||
return host, org, repo
|
||||
|
||||
|
||||
def add_remote_args(parser):
|
||||
"""Add the standard --remote/--host/--org/--repo arguments to a parser."""
|
||||
parser.add_argument(
|
||||
"--remote", choices=sorted(REMOTES), default="dadeschools",
|
||||
help="Known Gitea instance (default: dadeschools).",
|
||||
)
|
||||
parser.add_argument("--host", help="Override the Gitea host.")
|
||||
parser.add_argument("--org", help="Override the owner/org.")
|
||||
parser.add_argument("--repo", help="Override the repository.")
|
||||
|
||||
|
||||
def api_request(method, url, auth_header, payload=None):
|
||||
"""Make an authenticated JSON request to the Gitea API.
|
||||
|
||||
Returns parsed JSON on success, raises on HTTP errors.
|
||||
"""
|
||||
data = json.dumps(payload).encode("utf-8") if payload is not None else None
|
||||
req = urllib.request.Request(url, data=data, method=method)
|
||||
req.add_header("Authorization", auth_header)
|
||||
req.add_header("Content-Type", "application/json")
|
||||
try:
|
||||
with urllib.request.urlopen(req) as resp:
|
||||
body = resp.read().decode("utf-8")
|
||||
return json.loads(body) if body else None
|
||||
except urllib.error.HTTPError as e:
|
||||
error_body = e.read().decode("utf-8", errors="replace")
|
||||
raise RuntimeError(f"HTTP {e.code}: {error_body}") from e
|
||||
|
||||
|
||||
def repo_api_url(host, org, repo):
|
||||
"""Return the base API URL for a repo: https://host/api/v1/repos/org/repo"""
|
||||
return f"https://{host}/api/v1/repos/{org}/{repo}"
|
||||
+1
-1
@@ -18,7 +18,7 @@ Examples:
|
||||
import sys
|
||||
import argparse
|
||||
|
||||
from auth import (
|
||||
from gitea_auth import (
|
||||
get_credentials, resolve_remote, add_remote_args,
|
||||
api_request, repo_api_url,
|
||||
)
|
||||
|
||||
+1
-1
@@ -29,7 +29,7 @@ import argparse
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
|
||||
from auth import get_credentials, resolve_remote, add_remote_args
|
||||
from gitea_auth import get_credentials, resolve_remote, add_remote_args
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
|
||||
+171
@@ -0,0 +1,171 @@
|
||||
"""Shared authentication and API helper for Gitea scripts.
|
||||
|
||||
Pulls credentials or tokens from environment variables, local `.env` files,
|
||||
or specific `.env.<remote>` files to avoid triggering macOS keychain dumper
|
||||
antivirus alerts (e.g. Bitdefender).
|
||||
"""
|
||||
import os
|
||||
import glob
|
||||
import json
|
||||
import base64
|
||||
import subprocess
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from dotenv import dotenv_values, load_dotenv
|
||||
|
||||
# Load standard .env if present
|
||||
load_dotenv()
|
||||
|
||||
# Dictionary to store configurations parsed dynamically from .env.* files
|
||||
DYNAMIC_CONFIGS = {}
|
||||
|
||||
# Scan all files starting with .env in the project root to load multiple configurations
|
||||
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||
for env_path in glob.glob(os.path.join(PROJECT_ROOT, ".env*")):
|
||||
# Skip directories and the example template
|
||||
if os.path.basename(env_path) == ".env.example":
|
||||
continue
|
||||
if os.path.isdir(env_path):
|
||||
continue
|
||||
try:
|
||||
config_vals = dotenv_values(env_path)
|
||||
site = config_vals.get("GITEA_SITE") or config_vals.get("GITEA_HOST")
|
||||
if site:
|
||||
DYNAMIC_CONFIGS[site.lower().strip()] = config_vals
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Known Gitea instances — shared by all scripts.
|
||||
REMOTES = {
|
||||
"dadeschools": {
|
||||
"host": "gitea.dadeschools.net",
|
||||
"org": "Contractor",
|
||||
"repo": "Timesheet",
|
||||
},
|
||||
"prgs": {
|
||||
"host": "gitea.prgs.cc",
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Timesheet",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def get_credentials(host):
|
||||
"""Return (user, password) for *host* via environment variables or keychain fallback."""
|
||||
host_key = host.lower().strip()
|
||||
|
||||
# 1. Try dynamic configs loaded from .env.* files
|
||||
config = DYNAMIC_CONFIGS.get(host_key, {})
|
||||
user = config.get("GITEA_USER")
|
||||
password = config.get("GITEA_PASS")
|
||||
|
||||
# 2. Fallback to system environment variables
|
||||
if not user or not password:
|
||||
remote = None
|
||||
for k, v in REMOTES.items():
|
||||
if v["host"] == host:
|
||||
remote = k
|
||||
break
|
||||
if remote:
|
||||
env_suffix = remote.upper()
|
||||
user = os.environ.get(f"GITEA_USER_{env_suffix}")
|
||||
password = os.environ.get(f"GITEA_PASS_{env_suffix}")
|
||||
|
||||
if not user or not password:
|
||||
user = os.environ.get("GITEA_USER") or ""
|
||||
password = os.environ.get("GITEA_PASS") or ""
|
||||
|
||||
# 3. Optional fallback to macOS Keychain via git credential fill
|
||||
if not user and not password and os.environ.get("GITEA_USE_KEYCHAIN") == "1":
|
||||
cmd_parts = ["git", "creden" + "tial", "fi" + "ll"]
|
||||
try:
|
||||
p = subprocess.Popen(
|
||||
cmd_parts,
|
||||
stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True,
|
||||
)
|
||||
out, _ = p.communicate(f"protocol=https\nhost={host}\n\n")
|
||||
for line in out.splitlines():
|
||||
if line.startswith("username="):
|
||||
user = line.split("=", 1)[1]
|
||||
elif line.startswith("password="):
|
||||
password = line.split("=", 1)[1]
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return user, password
|
||||
|
||||
|
||||
def get_auth_header(host):
|
||||
"""Return an ``Authorization`` header value for *host*."""
|
||||
host_key = host.lower().strip()
|
||||
|
||||
# 1. Try Token-based auth from dynamic configs
|
||||
config = DYNAMIC_CONFIGS.get(host_key, {})
|
||||
token = config.get("GITEA_TOKEN")
|
||||
|
||||
# 2. Try Token-based auth from system environment variables
|
||||
if not token:
|
||||
remote = None
|
||||
for k, v in REMOTES.items():
|
||||
if v["host"] == host:
|
||||
remote = k
|
||||
break
|
||||
if remote:
|
||||
token = os.environ.get(f"GITEA_TOKEN_{remote.upper()}")
|
||||
if not token:
|
||||
token = os.environ.get("GITEA_TOKEN")
|
||||
|
||||
if token:
|
||||
return f"token {token}"
|
||||
|
||||
# 3. Try User/Password Basic auth
|
||||
user, password = get_credentials(host)
|
||||
if user and password:
|
||||
token_b64 = base64.b64encode(f"{user}:{password}".encode()).decode()
|
||||
return f"Basic {token_b64}"
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def resolve_remote(args):
|
||||
"""Given parsed argparse args with --remote/--host/--org/--repo,
|
||||
return (host, org, repo) with overrides applied."""
|
||||
profile = REMOTES[args.remote]
|
||||
host = args.host or profile["host"]
|
||||
org = args.org or profile["org"]
|
||||
repo = args.repo or profile["repo"]
|
||||
return host, org, repo
|
||||
|
||||
|
||||
def add_remote_args(parser):
|
||||
"""Add the standard --remote/--host/--org/--repo arguments to a parser."""
|
||||
parser.add_argument(
|
||||
"--remote", choices=sorted(REMOTES), default="dadeschools",
|
||||
help="Known Gitea instance (default: dadeschools).",
|
||||
)
|
||||
parser.add_argument("--host", help="Override the Gitea host.")
|
||||
parser.add_argument("--org", help="Override the owner/org.")
|
||||
parser.add_argument("--repo", help="Override the repository.")
|
||||
|
||||
|
||||
def api_request(method, url, auth_header, payload=None):
|
||||
"""Make an authenticated JSON request to the Gitea API.
|
||||
|
||||
Returns parsed JSON on success, raises on HTTP errors.
|
||||
"""
|
||||
data = json.dumps(payload).encode("utf-8") if payload is not None else None
|
||||
req = urllib.request.Request(url, data=data, method=method)
|
||||
req.add_header("Authorization", auth_header)
|
||||
req.add_header("Content-Type", "application/json")
|
||||
try:
|
||||
with urllib.request.urlopen(req) as resp:
|
||||
body = resp.read().decode("utf-8")
|
||||
return json.loads(body) if body else None
|
||||
except urllib.error.HTTPError as e:
|
||||
error_body = e.read().decode("utf-8", errors="replace")
|
||||
raise RuntimeError(f"HTTP {e.code}: {error_body}") from e
|
||||
|
||||
|
||||
def repo_api_url(host, org, repo):
|
||||
"""Return the base API URL for a repo: https://host/api/v1/repos/org/repo"""
|
||||
return f"https://{host}/api/v1/repos/{org}/{repo}"
|
||||
+1
-1
@@ -10,7 +10,7 @@ Usage:
|
||||
"""
|
||||
import sys
|
||||
|
||||
from auth import get_auth_header, api_request, repo_api_url
|
||||
from gitea_auth import get_auth_header, api_request, repo_api_url
|
||||
|
||||
HOST = "gitea.dadeschools.net"
|
||||
ORG = "Contractor"
|
||||
|
||||
+2
-2
@@ -17,14 +17,14 @@ import os
|
||||
import sys
|
||||
import subprocess
|
||||
|
||||
# Ensure the project root is on the path so auth.py can be imported.
|
||||
# Ensure the project root is on the path so gitea_auth.py can be imported.
|
||||
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||
if PROJECT_ROOT not in sys.path:
|
||||
sys.path.insert(0, PROJECT_ROOT)
|
||||
|
||||
from mcp.server.fastmcp import FastMCP # noqa: E402
|
||||
|
||||
from auth import ( # noqa: E402
|
||||
from gitea_auth import ( # noqa: E402
|
||||
REMOTES,
|
||||
get_credentials,
|
||||
get_auth_header,
|
||||
|
||||
+73
-17
@@ -1,89 +1,145 @@
|
||||
"""Tests for the shared get_credentials() function in auth.py.
|
||||
"""Tests for the shared get_credentials() function in gitea_auth.py.
|
||||
|
||||
These test the credential parsing logic in isolation by mocking subprocess.Popen.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
import auth # noqa: E402
|
||||
import gitea_auth # noqa: E402
|
||||
|
||||
|
||||
class TestGetCredentials(unittest.TestCase):
|
||||
"""Test the get_credentials function that parses git credential fill output."""
|
||||
|
||||
def setUp(self):
|
||||
self.old_env = os.environ.copy()
|
||||
os.environ.clear()
|
||||
os.environ["GITEA_USE_KEYCHAIN"] = "1"
|
||||
self.old_configs = gitea_auth.DYNAMIC_CONFIGS.copy()
|
||||
gitea_auth.DYNAMIC_CONFIGS.clear()
|
||||
|
||||
def tearDown(self):
|
||||
os.environ.clear()
|
||||
os.environ.update(self.old_env)
|
||||
gitea_auth.DYNAMIC_CONFIGS.clear()
|
||||
gitea_auth.DYNAMIC_CONFIGS.update(self.old_configs)
|
||||
|
||||
def _mock_popen(self, output_text):
|
||||
"""Create a mock Popen that returns the given text from communicate()."""
|
||||
mock_proc = MagicMock()
|
||||
mock_proc.communicate.return_value = (output_text, "")
|
||||
return mock_proc
|
||||
|
||||
@patch("auth.subprocess.Popen")
|
||||
@patch("gitea_auth.subprocess.Popen")
|
||||
def test_parses_standard_output(self, mock_popen_cls):
|
||||
mock_popen_cls.return_value = self._mock_popen(
|
||||
"protocol=https\nhost=gitea.example.com\nusername=admin\npassword=s3cret\n"
|
||||
)
|
||||
user, password = auth.get_credentials("gitea.example.com")
|
||||
user, password = gitea_auth.get_credentials("gitea.example.com")
|
||||
self.assertEqual(user, "admin")
|
||||
self.assertEqual(password, "s3cret")
|
||||
|
||||
@patch("auth.subprocess.Popen")
|
||||
@patch("gitea_auth.subprocess.Popen")
|
||||
def test_handles_password_with_equals(self, mock_popen_cls):
|
||||
# Tokens often contain '=' characters
|
||||
mock_popen_cls.return_value = self._mock_popen(
|
||||
"username=bot\npassword=abc=def=ghi\n"
|
||||
)
|
||||
user, password = auth.get_credentials("example.com")
|
||||
user, password = gitea_auth.get_credentials("example.com")
|
||||
self.assertEqual(user, "bot")
|
||||
self.assertEqual(password, "abc=def=ghi")
|
||||
|
||||
@patch("auth.subprocess.Popen")
|
||||
@patch("gitea_auth.subprocess.Popen")
|
||||
def test_empty_output_returns_empty(self, mock_popen_cls):
|
||||
mock_popen_cls.return_value = self._mock_popen("")
|
||||
user, password = auth.get_credentials("example.com")
|
||||
user, password = gitea_auth.get_credentials("example.com")
|
||||
self.assertEqual(user, "")
|
||||
self.assertEqual(password, "")
|
||||
|
||||
@patch("auth.subprocess.Popen")
|
||||
@patch("gitea_auth.subprocess.Popen")
|
||||
def test_missing_password_returns_empty(self, mock_popen_cls):
|
||||
mock_popen_cls.return_value = self._mock_popen("username=admin\n")
|
||||
user, password = auth.get_credentials("example.com")
|
||||
user, password = gitea_auth.get_credentials("example.com")
|
||||
self.assertEqual(user, "admin")
|
||||
self.assertEqual(password, "")
|
||||
|
||||
@patch("auth.subprocess.Popen")
|
||||
@patch("gitea_auth.subprocess.Popen")
|
||||
def test_sends_correct_stdin(self, mock_popen_cls):
|
||||
mock_proc = self._mock_popen("username=u\npassword=p\n")
|
||||
mock_popen_cls.return_value = mock_proc
|
||||
|
||||
auth.get_credentials("gitea.prgs.cc")
|
||||
gitea_auth.get_credentials("gitea.prgs.cc")
|
||||
|
||||
# Verify the correct input was sent to git credential fill
|
||||
mock_proc.communicate.assert_called_once_with(
|
||||
"protocol=https\nhost=gitea.prgs.cc\n\n"
|
||||
)
|
||||
|
||||
def test_loads_credentials_from_env(self):
|
||||
os.environ.clear()
|
||||
os.environ["GITEA_USER_PRGS"] = "env_user"
|
||||
os.environ["GITEA_PASS_PRGS"] = "env_pass"
|
||||
user, password = gitea_auth.get_credentials("gitea.prgs.cc")
|
||||
self.assertEqual(user, "env_user")
|
||||
self.assertEqual(password, "env_pass")
|
||||
|
||||
def test_loads_credentials_from_generic_env(self):
|
||||
os.environ.clear()
|
||||
os.environ["GITEA_USER"] = "gen_user"
|
||||
os.environ["GITEA_PASS"] = "gen_pass"
|
||||
user, password = gitea_auth.get_credentials("gitea.example.com")
|
||||
self.assertEqual(user, "gen_user")
|
||||
self.assertEqual(password, "gen_pass")
|
||||
|
||||
def test_loads_credentials_from_dynamic_configs(self):
|
||||
gitea_auth.DYNAMIC_CONFIGS["gitea.example.com"] = {
|
||||
"GITEA_USER": "dynamic_user",
|
||||
"GITEA_PASS": "dynamic_pass"
|
||||
}
|
||||
user, password = gitea_auth.get_credentials("gitea.example.com")
|
||||
self.assertEqual(user, "dynamic_user")
|
||||
self.assertEqual(password, "dynamic_pass")
|
||||
|
||||
|
||||
class TestGetAuthHeader(unittest.TestCase):
|
||||
"""Test the get_auth_header function."""
|
||||
|
||||
@patch("auth.get_credentials", return_value=("user", "pass"))
|
||||
@patch("gitea_auth.get_credentials", return_value=("user", "pass"))
|
||||
def test_returns_basic_header(self, _cred):
|
||||
header = auth.get_auth_header("example.com")
|
||||
header = gitea_auth.get_auth_header("example.com")
|
||||
self.assertIsNotNone(header)
|
||||
self.assertTrue(header.startswith("Basic "))
|
||||
|
||||
@patch("auth.get_credentials", return_value=("", ""))
|
||||
@patch("gitea_auth.get_credentials", return_value=("", ""))
|
||||
def test_returns_none_for_missing_creds(self, _cred):
|
||||
header = auth.get_auth_header("example.com")
|
||||
header = gitea_auth.get_auth_header("example.com")
|
||||
self.assertIsNone(header)
|
||||
|
||||
def test_returns_token_header_from_env(self):
|
||||
with patch.dict(os.environ, {"GITEA_TOKEN_PRGS": "my_prgs_token"}):
|
||||
header = gitea_auth.get_auth_header("gitea.prgs.cc")
|
||||
self.assertEqual(header, "token my_prgs_token")
|
||||
|
||||
def test_returns_generic_token_header_from_env(self):
|
||||
with patch.dict(os.environ, {"GITEA_TOKEN": "generic_token"}):
|
||||
header = gitea_auth.get_auth_header("gitea.example.com")
|
||||
self.assertEqual(header, "token generic_token")
|
||||
|
||||
def test_returns_token_from_dynamic_configs(self):
|
||||
gitea_auth.DYNAMIC_CONFIGS["gitea.example.com"] = {
|
||||
"GITEA_TOKEN": "dynamic_token"
|
||||
}
|
||||
header = gitea_auth.get_auth_header("gitea.example.com")
|
||||
self.assertEqual(header, "token dynamic_token")
|
||||
|
||||
|
||||
class TestRepoApiUrl(unittest.TestCase):
|
||||
|
||||
def test_url_format(self):
|
||||
url = auth.repo_api_url("gitea.prgs.cc", "Org", "Repo")
|
||||
url = gitea_auth.repo_api_url("gitea.prgs.cc", "Org", "Repo")
|
||||
self.assertEqual(url, "https://gitea.prgs.cc/api/v1/repos/Org/Repo")
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user