fix: workspace creation SSH authentication
- GitService.clone() now accepts ssh_key and sets up GIT_SSH_COMMAND env - WorkspaceManager.create() loads repo SSH key from DB and decrypts it - Both workspace create endpoints pass session for SSH key lookup Quality gates: ruff clean, pytest workspaces API (9 passed, 1 skipped)
This commit is contained in:
@@ -2,7 +2,9 @@
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -11,13 +13,39 @@ class GitService:
|
||||
"""Low-level git operations for creating and syncing workspaces."""
|
||||
|
||||
@staticmethod
|
||||
async def clone(remote_url: str, branch: str, path: str) -> None:
|
||||
def _prepare_ssh_env(
|
||||
ssh_key: str | None,
|
||||
) -> tuple[dict[str, str] | None, str | None]:
|
||||
"""Prepare environment for git commands with SSH authentication.
|
||||
|
||||
Returns a tuple of (env_dict, temp_key_path). Caller must clean up key_path.
|
||||
"""
|
||||
if not ssh_key:
|
||||
return None, None
|
||||
|
||||
fd, key_path = tempfile.mkstemp(prefix="ssh_key_")
|
||||
try:
|
||||
os.write(fd, ssh_key.encode())
|
||||
finally:
|
||||
os.close(fd)
|
||||
os.chmod(key_path, 0o600)
|
||||
|
||||
env = {
|
||||
"GIT_SSH_COMMAND": f"ssh -i {key_path} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null"
|
||||
}
|
||||
return env, key_path
|
||||
|
||||
@staticmethod
|
||||
async def clone(
|
||||
remote_url: str, branch: str, path: str, ssh_key: str | None = None
|
||||
) -> None:
|
||||
"""Clone a repository to the given path.
|
||||
|
||||
Args:
|
||||
remote_url: The git remote URL.
|
||||
branch: The branch to clone.
|
||||
path: The destination path for the clone.
|
||||
ssh_key: Optional decrypted SSH private key for authentication.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If the clone fails.
|
||||
@@ -31,17 +59,24 @@ class GitService:
|
||||
remote_url,
|
||||
path,
|
||||
]
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
stdout, stderr = await proc.communicate()
|
||||
if proc.returncode != 0:
|
||||
error_msg = stderr.decode().strip() if stderr else "unknown error"
|
||||
logger.error("Git clone failed: %s", error_msg)
|
||||
raise RuntimeError(f"Git clone failed: {error_msg}")
|
||||
logger.debug("Cloned %s (branch: %s) to %s", remote_url, branch, path)
|
||||
|
||||
env, key_path = GitService._prepare_ssh_env(ssh_key)
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
env={**os.environ, **env} if env else None,
|
||||
)
|
||||
stdout, stderr = await proc.communicate()
|
||||
if proc.returncode != 0:
|
||||
error_msg = stderr.decode().strip() if stderr else "unknown error"
|
||||
logger.error("Git clone failed: %s", error_msg)
|
||||
raise RuntimeError(f"Git clone failed: {error_msg}")
|
||||
logger.debug("Cloned %s (branch: %s) to %s", remote_url, branch, path)
|
||||
finally:
|
||||
if key_path and os.path.exists(key_path):
|
||||
os.unlink(key_path)
|
||||
|
||||
@staticmethod
|
||||
async def fetch(path: str) -> None:
|
||||
|
||||
Reference in New Issue
Block a user