diff --git a/Readme.md b/Readme.md index 58733ed..227a864 100644 --- a/Readme.md +++ b/Readme.md @@ -166,6 +166,19 @@ keychecker --version | **Codeberg** | `git@codeberg.org` | Username extraction | SaaS based on forgejo | | **Gitea** | `git@gitea.com` | Username extraction | Saas based on Gitea | | **Hugging Face** | `git@hf.co` | Username extraction | AI/ML model hosting platform | +| **DataOps.live** | `git@app.dataops.live` | Key confirmation | Data engineering platform | +| **Assembla** | `git@git.assembla.com` | Username extraction | Source code & collaboration platform | +| **Boltic** | `git@ssh.git.boltic.io` | Key confirmation | Serverless platform (Fynd Boltic) | +| **SourceHut** | `git@git.sr.ht` | Username extraction | Open source forge | +| **NotABug** | `git@notabug.org` | Username extraction | Gitea-based free code hosting | +| **Azure DevOps** | `git@ssh.dev.azure.com` | Key confirmation | Microsoft DevOps platform | +| **Framagit** | `git@framagit.org` | Username extraction | GitLab-based forge by Framasoft | +| **GitVerse** | `git@gitverse.ru` | Key confirmation | Russian Git platform by SberTech | +| **Launchpad** | `git@git.launchpad.net` | Key confirmation | Canonical/Ubuntu code hosting | +| **Gitee** πŸ‡¨πŸ‡³ | `git@gitee.com` | Username extraction | Chinese Git platform (OSCHINA) | +| **CODING.net** πŸ‡¨πŸ‡³ | `git@e.coding.net` | Key confirmation | Chinese DevOps platform (Tencent) | +| **CodeUP** πŸ‡¨πŸ‡³ | `git@codeup.aliyun.com` | Username extraction | Chinese DevOps Git (Alibaba Cloud) | +| **GitFlic** πŸ‡·πŸ‡Ί | `git@gitflic.ru` | Key confirmation | Russian Git hosting platform | --- diff --git a/keychecker/cli.py b/keychecker/cli.py index 15eafbe..dc9b921 100644 --- a/keychecker/cli.py +++ b/keychecker/cli.py @@ -55,9 +55,14 @@ def create_parser() -> argparse.ArgumentParser: parser.add_argument( "--validate", nargs="*", - choices=["github", "gitlab", "bitbucket", "codeberg", "gitea", "huggingface"], + choices=["github", "gitlab", "bitbucket", "codeberg", "gitea", "huggingface", + "dataops", "assembla", "boltic", "sourcehut", "notabug", + "azuredevops", "framagit", "gitverse", "launchpad", + "gitee", "coding", "codeup", "gitflic"], help=( - "One or more servers to validate against (default: all supported servers). " + "One or more servers to validate against (default: core providers). " + "Optional/regional providers: gitee(chinese), coding(chinese), " + "codeup(chinese), gitflic(russian). " "When used with --discover-repos, specifies which server to use for " "repository discovery." ), diff --git a/keychecker/core/server_validator.py b/keychecker/core/server_validator.py index 2b0a878..44d7717 100644 --- a/keychecker/core/server_validator.py +++ b/keychecker/core/server_validator.py @@ -12,6 +12,19 @@ CodebergProvider, GiteaProvider, HuggingFaceProvider, + AssemblaProvider, + AzureDevOpsProvider, + BolticProvider, + DataOpsProvider, + FramagitProvider, + GitVerseProvider, + LaunchpadProvider, + NotABugProvider, + SourceHutProvider, + CodingProvider, + CodeupProvider, + GiteeProvider, + GitFlicProvider, ) @@ -51,6 +64,45 @@ def __init__( "huggingface": HuggingFaceProvider( timeout=timeout, concurrency=concurrency, show_progress=show_progress ), + "dataops": DataOpsProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "assembla": AssemblaProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "boltic": BolticProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "sourcehut": SourceHutProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "notabug": NotABugProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "azuredevops": AzureDevOpsProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "framagit": FramagitProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "gitverse": GitVerseProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "launchpad": LaunchpadProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "coding": CodingProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "codeup": CodeupProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "gitee": GiteeProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), + "gitflic": GitFlicProvider( + timeout=timeout, concurrency=concurrency, show_progress=show_progress + ), } def get_supported_servers(self) -> List[str]: diff --git a/keychecker/plugins/__init__.py b/keychecker/plugins/__init__.py index 82db1c9..859f239 100644 --- a/keychecker/plugins/__init__.py +++ b/keychecker/plugins/__init__.py @@ -9,6 +9,19 @@ from .codeberg import CodebergProvider from .gitea import GiteaProvider from .huggingface import HuggingFaceProvider +from .assembla import AssemblaProvider +from .azuredevops import AzureDevOpsProvider +from .boltic import BolticProvider +from .dataops import DataOpsProvider +from .framagit import FramagitProvider +from .gitverse import GitVerseProvider +from .launchpad import LaunchpadProvider +from .notabug import NotABugProvider +from .sourcehut import SourceHutProvider +from .coding import CodingProvider +from .codeup import CodeupProvider +from .gitee import GiteeProvider +from .gitflic import GitFlicProvider __all__ = [ "BaseGitProvider", @@ -19,4 +32,17 @@ "CodebergProvider", "GiteaProvider", "HuggingFaceProvider", + "AssemblaProvider", + "AzureDevOpsProvider", + "BolticProvider", + "DataOpsProvider", + "FramagitProvider", + "GitVerseProvider", + "LaunchpadProvider", + "NotABugProvider", + "SourceHutProvider", + "CodingProvider", + "CodeupProvider", + "GiteeProvider", + "GitFlicProvider", ] diff --git a/keychecker/plugins/assembla.py b/keychecker/plugins/assembla.py new file mode 100644 index 0000000..d7f6905 --- /dev/null +++ b/keychecker/plugins/assembla.py @@ -0,0 +1,106 @@ +""" +Assembla-specific provider plugin for KeyChecker. + +Assembla is a source code hosting and collaboration platform. +SSH host: git.assembla.com (user: Git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class AssemblaProvider(BaseGitProvider): + """Assembla provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize Assembla provider.""" + config = ServerConfig( + name="assembla", hostname="git.assembla.com", port=22, username="Git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against Assembla and extract user information. + + Assembla returns specific banner format with username. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for Assembla success patterns + if "successfully authenticated" in output_lower: + result["authenticated"] = True + # Try to extract username from banner + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the Assembla username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover Assembla organizations. + + TODO: Implement Assembla organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test Assembla repository access. + """ + repo_url = f"git@git.assembla.com:{owner}/{repo_name}.git" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/azuredevops.py b/keychecker/plugins/azuredevops.py new file mode 100644 index 0000000..f84daae --- /dev/null +++ b/keychecker/plugins/azuredevops.py @@ -0,0 +1,107 @@ +""" +Azure DevOps-specific provider plugin for KeyChecker. + +Azure DevOps is Microsoft's DevOps platform with Git repository hosting. +SSH host: ssh.dev.azure.com (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class AzureDevOpsProvider(BaseGitProvider): + """Azure DevOps provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize Azure DevOps provider.""" + config = ServerConfig( + name="azuredevops", hostname="ssh.dev.azure.com", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against Azure DevOps and extract user information. + + Azure DevOps returns specific banner with authentication status. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for Azure DevOps success patterns + if "successfully authenticated" in output_lower: + result["authenticated"] = True + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the Azure DevOps username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover Azure DevOps organizations. + + TODO: Implement Azure DevOps organization discovery (via REST API). + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test Azure DevOps repository access. + + Azure DevOps SSH URLs follow: git@ssh.dev.azure.com:v3/{org}/{project}/{repo} + """ + repo_url = f"git@ssh.dev.azure.com:v3/{owner}/{repo_name}" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/boltic.py b/keychecker/plugins/boltic.py new file mode 100644 index 0000000..50d2262 --- /dev/null +++ b/keychecker/plugins/boltic.py @@ -0,0 +1,106 @@ +""" +Boltic (Fynd Boltic)-specific provider plugin for KeyChecker. + +Fynd Boltic is a serverless platform with Git integration. +SSH host: ssh.git.boltic.io (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class BolticProvider(BaseGitProvider): + """Boltic provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize Boltic provider.""" + config = ServerConfig( + name="boltic", hostname="ssh.git.boltic.io", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against Boltic and extract user information. + + Boltic returns standard SSH authentication banners. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for Boltic success patterns + if "successfully authenticated" in output_lower: + result["authenticated"] = True + # Try to extract username from banner + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the Boltic username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover Boltic organizations. + + TODO: Implement Boltic organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test Boltic repository access. + """ + repo_url = f"git@ssh.git.boltic.io:{owner}/{repo_name}.git" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/codeup.py b/keychecker/plugins/codeup.py new file mode 100644 index 0000000..42fe02e --- /dev/null +++ b/keychecker/plugins/codeup.py @@ -0,0 +1,112 @@ +""" +CodeUP (Alibaba Cloud)-specific provider plugin for KeyChecker. + +CodeUP is Alibaba Cloud's DevOps Git hosting platform (part of Yunxiao). +SSH host: codeup.aliyun.com (user: git) +Banner: "Welcome to Codeup, !" +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class CodeupProvider(BaseGitProvider): + """CodeUP provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize CodeUP provider.""" + config = ServerConfig( + name="codeup", hostname="codeup.aliyun.com", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against CodeUP and extract user information. + + CodeUP returns "Welcome to Codeup, !" banner on auth success. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for CodeUP success patterns + if "welcome to codeup" in output_lower: + result["authenticated"] = True + # Extract username from "Welcome to Codeup, !" + match = re.search(r"Welcome to Codeup, (.+?)(?:!|$)", output, re.IGNORECASE) + if match: + result["username"] = match.group(1).strip() + elif "successfully authenticated" in output_lower: + result["authenticated"] = True + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the CodeUP username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover CodeUP organizations. + + TODO: Implement CodeUP organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test CodeUP repository access. + """ + repo_url = f"git@codeup.aliyun.com:{owner}/{repo_name}.git" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/coding.py b/keychecker/plugins/coding.py new file mode 100644 index 0000000..65095f1 --- /dev/null +++ b/keychecker/plugins/coding.py @@ -0,0 +1,105 @@ +""" +CODING.net (Tencent)-specific provider plugin for KeyChecker. + +CODING.net is Tencent's DevOps platform with Git repository hosting. +SSH host: e.coding.net (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class CodingProvider(BaseGitProvider): + """CODING.net provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize CODING provider.""" + config = ServerConfig( + name="coding", hostname="e.coding.net", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against CODING.net and extract user information. + + CODING returns standard SSH authentication banners. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for CODING.net success patterns + if "successfully authenticated" in output_lower: + result["authenticated"] = True + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the CODING.net username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover CODING.net organizations. + + TODO: Implement CODING.net organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test CODING.net repository access. + """ + repo_url = f"git@e.coding.net:{owner}/{repo_name}.git" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/dataops.py b/keychecker/plugins/dataops.py new file mode 100644 index 0000000..a9a37dd --- /dev/null +++ b/keychecker/plugins/dataops.py @@ -0,0 +1,106 @@ +""" +DataOps.live-specific provider plugin for KeyChecker. + +DataOps.live is a data engineering platform with Git integration. +SSH host: app.dataops.live (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class DataOpsProvider(BaseGitProvider): + """DataOps.live provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize DataOps provider.""" + config = ServerConfig( + name="dataops", hostname="app.dataops.live", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against DataOps and extract user information. + + DataOps returns standard SSH authentication banners. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for DataOps success patterns + if "successfully authenticated" in output_lower: + result["authenticated"] = True + # Try to extract username from banner + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the DataOps username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover DataOps organizations. + + TODO: Implement DataOps organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test DataOps repository access. + """ + repo_url = f"git@app.dataops.live:{owner}/{repo_name}.git" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/framagit.py b/keychecker/plugins/framagit.py new file mode 100644 index 0000000..e12144c --- /dev/null +++ b/keychecker/plugins/framagit.py @@ -0,0 +1,110 @@ +""" +Framagit-specific provider plugin for KeyChecker. + +Framagit is a GitLab-based software forge hosted by Framasoft. +SSH host: framagit.org (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class FramagitProvider(BaseGitProvider): + """Framagit provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize Framagit provider.""" + config = ServerConfig( + name="framagit", hostname="framagit.org", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against Framagit and extract user information. + + Framagit is GitLab-based so it returns GitLab-style banners. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for Framagit success patterns (GitLab-based) + if "welcome to gitlab" in output_lower or "successfully authenticated" in output_lower: + result["authenticated"] = True + # GitLab format: "Welcome to GitLab, @username!" + match = re.search(r"Welcome to GitLab, @?(\w+)", output, re.IGNORECASE) + if match: + result["username"] = match.group(1) + else: + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the Framagit username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover Framagit organizations/groups. + + TODO: Implement Framagit organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test Framagit repository access. + """ + repo_url = f"git@framagit.org:{owner}/{repo_name}.git" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/gitee.py b/keychecker/plugins/gitee.py new file mode 100644 index 0000000..da61bc5 --- /dev/null +++ b/keychecker/plugins/gitee.py @@ -0,0 +1,110 @@ +""" +Gitee (码云)-specific provider plugin for KeyChecker. + +Gitee is China's largest Git hosting platform, operated by OSCHINA. +SSH host: gitee.com (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class GiteeProvider(BaseGitProvider): + """Gitee provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize Gitee provider.""" + config = ServerConfig( + name="gitee", hostname="gitee.com", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against Gitee and extract user information. + + Gitee returns GitHub-like "Hi !" banner on auth success. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for Gitee success patterns + if "successfully authenticated" in output_lower: + result["authenticated"] = True + # Try "Hi !" pattern (GitHub-like) + match = re.search(r"Hi (.+)!", output, re.IGNORECASE) + if match: + result["username"] = match.group(1).strip() + else: + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the Gitee username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover Gitee organizations. + + TODO: Implement Gitee organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test Gitee repository access. + """ + repo_url = f"git@gitee.com:{owner}/{repo_name}.git" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/gitflic.py b/keychecker/plugins/gitflic.py new file mode 100644 index 0000000..5878653 --- /dev/null +++ b/keychecker/plugins/gitflic.py @@ -0,0 +1,105 @@ +""" +GitFlic-specific provider plugin for KeyChecker. + +GitFlic is a Russian Git hosting and DevSecOps platform. +SSH host: gitflic.ru (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class GitFlicProvider(BaseGitProvider): + """GitFlic provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize GitFlic provider.""" + config = ServerConfig( + name="gitflic", hostname="gitflic.ru", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against GitFlic and extract user information. + + GitFlic returns standard SSH authentication banners. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for GitFlic success patterns + if "successfully authenticated" in output_lower: + result["authenticated"] = True + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the GitFlic username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover GitFlic organizations. + + TODO: Implement GitFlic organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test GitFlic repository access. + """ + repo_url = f"git@gitflic.ru:{owner}/{repo_name}.git" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/gitverse.py b/keychecker/plugins/gitverse.py new file mode 100644 index 0000000..24ce4df --- /dev/null +++ b/keychecker/plugins/gitverse.py @@ -0,0 +1,105 @@ +""" +GitVerse-specific provider plugin for KeyChecker. + +GitVerse is a Russian Git hosting platform by SberTech. +SSH host: gitverse.ru (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class GitVerseProvider(BaseGitProvider): + """GitVerse provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize GitVerse provider.""" + config = ServerConfig( + name="gitverse", hostname="gitverse.ru", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against GitVerse and extract user information. + + GitVerse returns standard SSH authentication banners. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for GitVerse success patterns + if "successfully authenticated" in output_lower: + result["authenticated"] = True + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the GitVerse username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover GitVerse organizations. + + TODO: Implement GitVerse organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test GitVerse repository access. + """ + repo_url = f"git@gitverse.ru:{owner}/{repo_name}.git" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/launchpad.py b/keychecker/plugins/launchpad.py new file mode 100644 index 0000000..ec91a02 --- /dev/null +++ b/keychecker/plugins/launchpad.py @@ -0,0 +1,105 @@ +""" +Launchpad-specific provider plugin for KeyChecker. + +Launchpad is Canonical/Ubuntu's platform for code hosting, now supporting Git. +SSH host: git.launchpad.net (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class LaunchpadProvider(BaseGitProvider): + """Launchpad provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize Launchpad provider.""" + config = ServerConfig( + name="launchpad", hostname="git.launchpad.net", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against Launchpad and extract user information. + + Launchpad returns standard SSH authentication banners. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for Launchpad success patterns + if "successfully authenticated" in output_lower: + result["authenticated"] = True + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the Launchpad username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover Launchpad organizations/teams. + + TODO: Implement Launchpad organization/team discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test Launchpad repository access. + """ + repo_url = f"git@git.launchpad.net:{owner}/{repo_name}" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/notabug.py b/keychecker/plugins/notabug.py new file mode 100644 index 0000000..c2b8912 --- /dev/null +++ b/keychecker/plugins/notabug.py @@ -0,0 +1,111 @@ +""" +NotABug-specific provider plugin for KeyChecker. + +NotABug.org is a free-software code collaboration platform based on Gitea. +SSH host: notabug.org (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class NotABugProvider(BaseGitProvider): + """NotABug provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize NotABug provider.""" + config = ServerConfig( + name="notabug", hostname="notabug.org", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against NotABug and extract user information. + + NotABug is Gitea-based, so it likely returns similar banners. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for NotABug success patterns (Gitea-based) + if "hi there," in output_lower and "successfully authenticated" in output_lower: + result["authenticated"] = True + # Extract username from "Hi there, username!" + match = re.search(r"Hi there, (.+)!", output, re.IGNORECASE) + if match: + result["username"] = match.group(1) + elif "successfully authenticated" in output_lower: + result["authenticated"] = True + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the NotABug username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover NotABug organizations. + + TODO: Implement NotABug organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test NotABug repository access. + """ + repo_url = f"git@notabug.org:{owner}/{repo_name}.git" + return await self._run_git_ls_remote(private_key_path, repo_url) diff --git a/keychecker/plugins/sourcehut.py b/keychecker/plugins/sourcehut.py new file mode 100644 index 0000000..6658b2c --- /dev/null +++ b/keychecker/plugins/sourcehut.py @@ -0,0 +1,112 @@ +""" +SourceHut (sr.ht)-specific provider plugin for KeyChecker. + +SourceHut is a minimalist open source forge with Git and Mercurial hosting. +SSH host: git.sr.ht (user: git) +""" + +from typing import Dict, Any, List, Optional +import re + +from .base import BaseGitProvider, ServerConfig + + +class SourceHutProvider(BaseGitProvider): + """SourceHut provider implementation.""" + + def __init__( + self, timeout: int = 5, concurrency: int = 10, show_progress: bool = True + ): + """Initialize SourceHut provider.""" + config = ServerConfig( + name="sourcehut", hostname="git.sr.ht", port=22, username="git" + ) + super().__init__(config, timeout, concurrency, show_progress) + + async def validate_key(self, private_key_path: str) -> Dict[str, Any]: + """ + Validate SSH key against SourceHut and extract user information. + + SourceHut returns "Welcome to sourcehut, !" on auth success. + """ + exit_code, stdout, stderr = await self._run_ssh_command(private_key_path) + + # Combine stdout and stderr for analysis + output = (stdout + stderr).strip() + output_lower = output.lower() + + result = { + "reachable": True, + "authenticated": False, + "username": None, + "banner": output, + "error": None, + } + + # Check for SourceHut success patterns + if "welcome to sourcehut" in output_lower: + result["authenticated"] = True + # Extract username from "Welcome to sourcehut, !" + match = re.search(r"Welcome to sourcehut, (.+)!", output, re.IGNORECASE) + if match: + result["username"] = match.group(1).strip() + elif "successfully authenticated" in output_lower: + result["authenticated"] = True + username = self._extract_username_from_banner(output) + if username: + result["username"] = username + elif any( + pattern in output_lower + for pattern in [ + "permission denied (publickey)", + "permission denied for user", + "authentication failed", + "publickey authentication failed", + ] + ): + result["authenticated"] = False + result["error"] = "Authentication failed - key not authorized" + elif any( + pattern in output_lower + for pattern in [ + "connection refused", + "connection timed out", + "network is unreachable", + ] + ): + result["reachable"] = False + result["error"] = ( + f'Connection failed - {output.split()[0] if output else "unknown"}' + ) + elif exit_code != 0: + result["authenticated"] = False + result["error"] = "Authentication failed - unknown error" + + return result + + async def identify_user(self, private_key_path: str) -> Optional[str]: + """ + Identify the SourceHut username associated with the private key. + """ + validation_result = await self.validate_key(private_key_path) + return validation_result.get("username") + + async def discover_organizations( + self, private_key_path: str, username: str + ) -> List[str]: + """ + Discover SourceHut organizations/namespaces. + + TODO: Implement SourceHut organization discovery. + """ + return [] + + async def test_repository_access( + self, private_key_path: str, owner: str, repo_name: str + ) -> bool: + """ + Test SourceHut repository access. + SourceHut uses ~owner/repo format. + """ + repo_url = f"git@git.sr.ht:~{owner}/{repo_name}" + return await self._run_git_ls_remote(private_key_path, repo_url)