Skip to content
Open
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
13 changes: 13 additions & 0 deletions Readme.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |

---

Expand Down
9 changes: 7 additions & 2 deletions keychecker/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."
),
Expand Down
52 changes: 52 additions & 0 deletions keychecker/core/server_validator.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,19 @@
CodebergProvider,
GiteaProvider,
HuggingFaceProvider,
AssemblaProvider,
AzureDevOpsProvider,
BolticProvider,
DataOpsProvider,
FramagitProvider,
GitVerseProvider,
LaunchpadProvider,
NotABugProvider,
SourceHutProvider,
CodingProvider,
CodeupProvider,
GiteeProvider,
GitFlicProvider,
)


Expand Down Expand Up @@ -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]:
Expand Down
26 changes: 26 additions & 0 deletions keychecker/plugins/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -19,4 +32,17 @@
"CodebergProvider",
"GiteaProvider",
"HuggingFaceProvider",
"AssemblaProvider",
"AzureDevOpsProvider",
"BolticProvider",
"DataOpsProvider",
"FramagitProvider",
"GitVerseProvider",
"LaunchpadProvider",
"NotABugProvider",
"SourceHutProvider",
"CodingProvider",
"CodeupProvider",
"GiteeProvider",
"GitFlicProvider",
]
106 changes: 106 additions & 0 deletions keychecker/plugins/assembla.py
Original file line number Diff line number Diff line change
@@ -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)
107 changes: 107 additions & 0 deletions keychecker/plugins/azuredevops.py
Original file line number Diff line number Diff line change
@@ -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)
Loading
Loading