fix usename exposure
This commit is contained in:
+33
-39
@@ -1,9 +1,12 @@
|
|||||||
import os
|
import os
|
||||||
import logging
|
import logging
|
||||||
import subprocess
|
import subprocess
|
||||||
|
import base64
|
||||||
|
import shutil
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from urllib.parse import urlparse
|
from urllib.parse import urlparse
|
||||||
from .config import GITEA_REPOS_ROOT, GITEA_URL, GITEA_TOKEN
|
from .config import GITEA_REPOS_ROOT, GITEA_URL, GITEA_TOKEN
|
||||||
|
from gitea.client import GiteaClient
|
||||||
|
|
||||||
logger: logging.Logger = logging.getLogger("gitea-workspace")
|
logger: logging.Logger = logging.getLogger("gitea-workspace")
|
||||||
|
|
||||||
@@ -14,49 +17,25 @@ class WorkspaceManager:
|
|||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
self.root_dir: Path = Path(GITEA_REPOS_ROOT).expanduser().resolve()
|
self.root_dir: Path = Path(GITEA_REPOS_ROOT).expanduser().resolve()
|
||||||
self.root_dir.mkdir(parents=True, exist_ok=True)
|
self.root_dir.mkdir(parents=True, exist_ok=True)
|
||||||
self._configure_git_credentials()
|
|
||||||
|
|
||||||
def _configure_git_credentials(self) -> None:
|
|
||||||
try:
|
|
||||||
# Unset any global configs we might have set previously
|
|
||||||
subprocess.run(
|
|
||||||
["git", "config", "--global", "--unset", "credential.helper"],
|
|
||||||
capture_output=True
|
|
||||||
)
|
|
||||||
subprocess.run(
|
|
||||||
["git", "config", "--global", "--unset", "user.name"],
|
|
||||||
capture_output=True
|
|
||||||
)
|
|
||||||
subprocess.run(
|
|
||||||
["git", "config", "--global", "--unset", "user.email"],
|
|
||||||
capture_output=True
|
|
||||||
)
|
|
||||||
except Exception as e:
|
|
||||||
logger.error(f"Error unsetting global configs: {e}")
|
|
||||||
|
|
||||||
def _configure_repo_user(self, repo_path: Path) -> None:
|
def _configure_repo_user(self, repo_path: Path) -> None:
|
||||||
try:
|
try:
|
||||||
# Configure credential helper locally for the repo
|
|
||||||
subprocess.run(
|
|
||||||
["git", "-C", str(repo_path), "config", "credential.helper", "store"],
|
|
||||||
check=True, capture_output=True
|
|
||||||
)
|
|
||||||
# Write to ~/.git-credentials
|
|
||||||
parsed = urlparse(GITEA_URL.rstrip("/"))
|
|
||||||
cred_line = f"{parsed.scheme}://meeks-ai:{GITEA_TOKEN}@{parsed.netloc}\n"
|
|
||||||
cred_file = Path("~/.git-credentials").expanduser()
|
|
||||||
if cred_file.exists():
|
|
||||||
content = cred_file.read_text()
|
|
||||||
if cred_line.strip() not in content:
|
|
||||||
cred_file.write_text(content + cred_line)
|
|
||||||
else:
|
|
||||||
cred_file.write_text(cred_line)
|
|
||||||
|
|
||||||
from gitea.client import GiteaClient
|
|
||||||
client = GiteaClient()
|
client = GiteaClient()
|
||||||
user = client.get_authenticated_user()
|
user = client.get_authenticated_user()
|
||||||
|
username = user.login if user else "unknown-ai"
|
||||||
|
|
||||||
|
auth_str = f"{username}:{GITEA_TOKEN}"
|
||||||
|
auth_bytes = auth_str.encode("utf-8")
|
||||||
|
auth_b64 = base64.b64encode(auth_bytes).decode("utf-8")
|
||||||
|
|
||||||
|
# Configure extraHeader locally for the repo
|
||||||
|
subprocess.run(
|
||||||
|
["git", "-C", str(repo_path), "config", "http.extraHeader", f"Authorization: Basic {auth_b64}"],
|
||||||
|
check=True, capture_output=True
|
||||||
|
)
|
||||||
|
|
||||||
if user:
|
if user:
|
||||||
name = user.full_name or user.login or "meeks-ai"
|
name = user.full_name or user.login or "unknown-ai"
|
||||||
email = user.email or f"{user.login or 'agent'}@noreply.gitea"
|
email = user.email or f"{user.login or 'agent'}@noreply.gitea"
|
||||||
subprocess.run(
|
subprocess.run(
|
||||||
["git", "-C", str(repo_path), "config", "user.name", name],
|
["git", "-C", str(repo_path), "config", "user.name", name],
|
||||||
@@ -123,13 +102,28 @@ class WorkspaceManager:
|
|||||||
if not (repo_path / ".git").exists():
|
if not (repo_path / ".git").exists():
|
||||||
new_path: Path = repo_path.parent / f"{repo_full_name.replace('/', '_')}_old"
|
new_path: Path = repo_path.parent / f"{repo_full_name.replace('/', '_')}_old"
|
||||||
if new_path.exists():
|
if new_path.exists():
|
||||||
import shutil
|
|
||||||
shutil.rmtree(new_path)
|
shutil.rmtree(new_path)
|
||||||
repo_path.rename(new_path)
|
repo_path.rename(new_path)
|
||||||
return repo_path
|
return repo_path
|
||||||
|
|
||||||
logger.info(f"Cloning repository {repo_full_name} to {repo_path}...")
|
logger.info(f"Cloning repository {repo_full_name} to {repo_path}...")
|
||||||
auth_url = self._get_authenticated_url(repo_full_name)
|
auth_url = self._get_authenticated_url(repo_full_name)
|
||||||
subprocess.run(["git", "clone", auth_url, str(repo_path)], check=True, capture_output=True)
|
|
||||||
|
username = "unknown-ai"
|
||||||
|
try:
|
||||||
|
client = GiteaClient()
|
||||||
|
user = client.get_authenticated_user()
|
||||||
|
if user:
|
||||||
|
username = user.login
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
auth_str = f"{username}:{GITEA_TOKEN}"
|
||||||
|
auth_bytes = auth_str.encode("utf-8")
|
||||||
|
auth_b64 = base64.b64encode(auth_bytes).decode("utf-8")
|
||||||
|
subprocess.run(
|
||||||
|
["git", "clone", "-c", f"http.extraHeader=Authorization: Basic {auth_b64}", auth_url, str(repo_path)],
|
||||||
|
check=True, capture_output=True
|
||||||
|
)
|
||||||
self._configure_repo_user(repo_path)
|
self._configure_repo_user(repo_path)
|
||||||
return repo_path
|
return repo_path
|
||||||
|
|||||||
@@ -0,0 +1,53 @@
|
|||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from gitea.workspace import WorkspaceManager
|
||||||
|
|
||||||
|
|
||||||
|
@patch("gitea.workspace.subprocess.run")
|
||||||
|
@patch("gitea.workspace.GiteaClient")
|
||||||
|
def test_workspace_manager_configure_repo_user(
|
||||||
|
mock_client_class: MagicMock,
|
||||||
|
mock_run: MagicMock
|
||||||
|
) -> None:
|
||||||
|
mock_client = MagicMock()
|
||||||
|
mock_client_class.return_value = mock_client
|
||||||
|
mock_user = MagicMock()
|
||||||
|
mock_user.full_name = "Agent Tester"
|
||||||
|
mock_user.login = "agent-test"
|
||||||
|
mock_user.email = "agent-test@example.com"
|
||||||
|
mock_client.get_authenticated_user.return_value = mock_user
|
||||||
|
|
||||||
|
workspace = WorkspaceManager()
|
||||||
|
repo_path = Path("/tmp/mock-repo")
|
||||||
|
|
||||||
|
workspace._configure_repo_user(repo_path)
|
||||||
|
|
||||||
|
assert mock_run.call_count >= 3
|
||||||
|
calls = [c[0][0] for c in mock_run.call_args_list]
|
||||||
|
|
||||||
|
assert any("http.extraHeader" in call for call in calls)
|
||||||
|
assert any("user.name" in call for call in calls)
|
||||||
|
assert any("user.email" in call for call in calls)
|
||||||
|
|
||||||
|
|
||||||
|
@patch("gitea.workspace.subprocess.run")
|
||||||
|
def test_workspace_manager_clone_repo(
|
||||||
|
mock_run: MagicMock
|
||||||
|
) -> None:
|
||||||
|
workspace = WorkspaceManager()
|
||||||
|
|
||||||
|
with patch.object(workspace, "_configure_repo_user") as mock_configure:
|
||||||
|
with patch.object(workspace, "get_repo_path") as mock_get_path:
|
||||||
|
mock_repo_path = MagicMock(spec=Path)
|
||||||
|
mock_repo_path.exists.return_value = False
|
||||||
|
mock_get_path.return_value = mock_repo_path
|
||||||
|
|
||||||
|
workspace.clone_repo("meeks/repo1")
|
||||||
|
|
||||||
|
mock_run.assert_called_once()
|
||||||
|
args = mock_run.call_args[0][0]
|
||||||
|
assert "clone" in args
|
||||||
|
assert any("http.extraHeader=Authorization: Basic" in arg for arg in args)
|
||||||
|
mock_configure.assert_called_once_with(mock_repo_path)
|
||||||
Reference in New Issue
Block a user