Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitattributes
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
modules/git_askpass.py text eol=lf
15 changes: 9 additions & 6 deletions git_platforms_synchro.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import modules.input_parser as input_parser
from git import Repo
from modules.git_clients import GitClientFactory, GitClient
from modules.utils import TMP_REPO_GIT_DIRECTORY, delete_temporary_repo_git_directory
from modules.utils import TMP_REPO_GIT_DIRECTORY, delete_temporary_repo_git_directory, test_git_ask_pass, get_git_ask_pass

GIT_CONFIG_HTTP_PREFIX = 'http'
GIT_REMOTE_TO = 'sync-to'
Expand All @@ -29,13 +29,15 @@ def set_git_credentials(username: str, password: str):
os.environ.pop('GIT_PASSWORD', None)
return

working_dir = os.path.dirname(os.path.realpath(__file__))

os.environ['GIT_ASKPASS'] = os.path.join(working_dir, 'modules', 'git_askpass.py')
os.environ['GIT_ASKPASS'] = get_git_ask_pass()
if username:
os.environ['GIT_USERNAME'] = username
else:
os.environ.pop('GIT_USERNAME', None)
if password:
os.environ['GIT_PASSWORD'] = password
else:
os.environ.pop('GIT_PASSWORD', None)


def git_clone(url: str, mirror: bool = False, disable_ssl_verify: bool = False, proxy: str = None) -> Repo:
Expand All @@ -49,7 +51,7 @@ def git_clone(url: str, mirror: bool = False, disable_ssl_verify: bool = False,
else:
delete_temporary_repo_git_directory()
logging.debug('Cloning repo %s', url)
options = []
options = ['--config credential.helper=""']
if disable_ssl_verify:
options += ['--config http.sslVerify=false']
if proxy:
Expand Down Expand Up @@ -146,9 +148,10 @@ def repo_branches_sync(args, branches_commits_from: dict, branches_commits_to: d


def main() -> int:
delete_temporary_repo_git_directory()
args = input_parser.parse()
delete_temporary_repo_git_directory()
log_init(args.log_level)
test_git_ask_pass()
logger.info('Starting Git Platforms Synchronization...')
input_parser.print_args(args)

Expand Down
29 changes: 29 additions & 0 deletions modules/git_askpass.bat
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
@echo off
:: Script to read GIT_USERNAME or GIT_PASSWORD from environment variables

:: Check if an argument is provided
if "%~1"=="" (
echo "Usage: %~nx0 [Username|Password]"
exit /b 1
)

:: Convert the argument to lowercase for comparison
set "param=%~1"
set "param_lower=%param:~0,1%%param:~1%"

:: Check if the argument contains "username"
echo %param_lower% | findstr /i "username" >nul
if %errorlevel% equ 0 (
echo %GIT_USERNAME%
exit /b 0
)

:: Check if the argument contains "password"
echo %param_lower% | findstr /i "password" >nul
if %errorlevel% equ 0 (
echo %GIT_PASSWORD%
exit /b 0
)

:: If none of the cases are found
exit /b 1
2 changes: 1 addition & 1 deletion modules/git_askpass.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
#!/usr/bin/python
#!/usr/bin/env python3

import sys
from sys import argv
Expand Down
32 changes: 32 additions & 0 deletions modules/utils.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import os
import stat
import shutil
import subprocess

Expand All @@ -18,3 +19,34 @@ def delete_temporary_repo_git_directory(force_if_test_mode: bool = False):
subprocess.run(['cmd', '/c', 'rmdir', '/s', '/q', TMP_REPO_GIT_DIRECTORY.rstrip('/')], shell=True)
else:
raise e


def set_file_execution_permission(file: str):
if not os.access(file, os.X_OK):
permissions = os.stat(file).st_mode
os.chmod(file, permissions | stat.S_IXUSR | stat.S_IXGRP)


def get_git_ask_pass() -> str:
os_windows = os.name == 'nt'
working_dir = os.path.dirname(os.path.realpath(__file__))
git_askpass = os.path.join(working_dir, 'git_askpass.' + ('bat' if os_windows else 'py'))
if not os_windows:
set_file_execution_permission(git_askpass)
return git_askpass


def test_git_ask_pass() -> None:
git_askpass = get_git_ask_pass()
custom_env = os.environ.copy()
custom_env['GIT_USERNAME'] = 'test42'
custom_env['GIT_PASSWORD'] = '42test' # noqa: S2068
result_user = subprocess.run([git_askpass, 'Username'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=custom_env)
result_pwd = subprocess.run([git_askpass, 'Password'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=custom_env)
os.environ.pop('GIT_USERNAME', None)
os.environ.pop('GIT_PASSWORD', None)
if result_user.returncode != 0 or result_user.stdout != 'test42\n' or result_pwd.returncode != 0 or result_pwd.stdout != '42test\n':
raise ValueError(
'PROBLEM: The ' +
git_askpass +
' used for Git authentications without credentials storage cannot be executed, please verify file Dos/Unix encoding and/or permissions.')
15 changes: 14 additions & 1 deletion tests/test_utils.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
import os
import json
import tarfile
import tempfile
import pytest
from unittest.mock import patch
from git import Repo
from modules.utils import ENV_TEST_MODE, TMP_REPO_GIT_DIRECTORY, delete_temporary_repo_git_directory
from modules.utils import ENV_TEST_MODE, TMP_REPO_GIT_DIRECTORY, delete_temporary_repo_git_directory, set_file_execution_permission, test_git_ask_pass
from pytest_httpserver import HTTPServer


Expand Down Expand Up @@ -74,3 +75,15 @@ def test_delete_temporary_repo_git_directory_permission_denied_windows(capsys):
with patch('subprocess.run', return_value=mock_result) as mock_run:
delete_temporary_repo_git_directory(True)
mock_run.assert_called_once_with(['cmd', '/c', 'rmdir', '/s', '/q', 'tmp-git-repo'], shell=True)


@pytest.mark.skipif(os.name == 'nt', reason='Specific Linux test')
def test_set_file_execution_permission():
with tempfile.NamedTemporaryFile(mode='w', delete=True) as temp_file:
assert not os.access(temp_file.name, os.X_OK)
set_file_execution_permission(temp_file.name)
assert os.access(temp_file.name, os.X_OK)


def test_test_git_ask_pass():
test_git_ask_pass()