forked from ci4s/pipeline-convert
git拉取私有代码支持ssh方式和用户名密码方式
This commit is contained in:
parent
7fe305eb00
commit
7330b30567
|
|
@ -47,6 +47,11 @@ func (r *refCodeTypeParser) ParseType(key string, param model.Param) (*TypeResul
|
|||
args = append(args, fmt.Sprintf("%s=%s", sshKey, codeParam.SshPrivateKey))
|
||||
}
|
||||
|
||||
if codeParam.Username != "" && codeParam.Password != "" {
|
||||
args = append(args, fmt.Sprintf("%s=%s", username, codeParam.Username),
|
||||
fmt.Sprintf("%s=%s", password, codeParam.Password))
|
||||
}
|
||||
|
||||
return &TypeResult{
|
||||
Args: args,
|
||||
}, nil
|
||||
|
|
|
|||
|
|
@ -4,28 +4,47 @@ import os
|
|||
import sys
|
||||
import tempfile
|
||||
|
||||
def clone_repo(ssh_key_content, repo_url, branch, depth, output_dir):
|
||||
def clone_repo(ssh_key_content, username, password, repo_url, branch, depth, output_dir):
|
||||
# 确保输出目录存在
|
||||
if not os.path.exists(output_dir):
|
||||
os.makedirs(output_dir)
|
||||
|
||||
# 仅当提供了 SSH 私钥内容时,写入临时文件
|
||||
# 如果提供了 SSH 私钥内容,使用 SSH 方式克隆代码
|
||||
if ssh_key_content:
|
||||
print("use ssh private key")
|
||||
print("Using SSH private key for cloning.")
|
||||
with tempfile.NamedTemporaryFile(delete=False) as tmpfile:
|
||||
ssh_key_path = tmpfile.name
|
||||
tmpfile.write(ssh_key_content.encode())
|
||||
print("ssh key path is ", ssh_key_path)
|
||||
print("SSH key path is:", ssh_key_path)
|
||||
os.chmod(ssh_key_path, 0o600)
|
||||
|
||||
# 设置 SSH 私钥的环境变量
|
||||
os.environ['GIT_SSH_COMMAND'] = f'ssh -i {ssh_key_path} -o StrictHostKeyChecking=no'
|
||||
|
||||
# 构建 git clone 命令
|
||||
clone_cmd = [
|
||||
'git', 'clone', '-b', branch,
|
||||
'--depth', str(depth), repo_url, output_dir
|
||||
]
|
||||
clone_cmd = [
|
||||
'git', 'clone', '-b', branch,
|
||||
'--depth', str(depth), repo_url, output_dir
|
||||
]
|
||||
|
||||
# 如果提供了用户名和密码,使用 HTTP/HTTPS 方式克隆代码
|
||||
elif username and password:
|
||||
print("Using username and password for cloning.")
|
||||
# 构造带有用户名和密码的 URL
|
||||
protocol, url_without_protocol = repo_url.split("://")
|
||||
auth_url = f"{protocol}://{username}:{password}@{url_without_protocol}"
|
||||
|
||||
clone_cmd = [
|
||||
'git', 'clone', '-b', branch,
|
||||
'--depth', str(depth), auth_url, output_dir
|
||||
]
|
||||
|
||||
# 如果既没有 SSH 私钥也没有用户名和密码,使用常规方式克隆公共仓库
|
||||
else:
|
||||
print("Cloning public repository without authentication.")
|
||||
clone_cmd = [
|
||||
'git', 'clone', '-b', branch,
|
||||
'--depth', str(depth), repo_url, output_dir
|
||||
]
|
||||
|
||||
# 执行 git clone 命令
|
||||
try:
|
||||
|
|
@ -39,7 +58,6 @@ def clone_repo(ssh_key_content, repo_url, branch, depth, output_dir):
|
|||
os.remove(ssh_key_path)
|
||||
|
||||
return True
|
||||
|
||||
def main():
|
||||
# 解析命令行参数
|
||||
parser = argparse.ArgumentParser(description='Clone a Git repository.')
|
||||
|
|
@ -47,6 +65,8 @@ def main():
|
|||
parser.add_argument('--code_path', required=True, help='Git repository URL')
|
||||
parser.add_argument('--branch', default='master', help='Branch to clone')
|
||||
parser.add_argument('--depth', type=int, default=1, help='Depth for shallow clone')
|
||||
parser.add_argument('--username', type=str, default="", help='username for git clone')
|
||||
parser.add_argument('--password', type=str, default="", help='password for git clone')
|
||||
parser.add_argument('--code_output', required=True, help='Directory to clone the repository into')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
|
@ -54,6 +74,8 @@ def main():
|
|||
# 执行克隆操作
|
||||
success = clone_repo(
|
||||
args.ssh_private_key,
|
||||
args.username,
|
||||
args.password,
|
||||
args.code_path,
|
||||
args.branch,
|
||||
args.depth,
|
||||
|
|
|
|||
Loading…
Reference in New Issue