Compare commits

17 Commits
Author SHA1 Message Date
ahp 72add0f0ac ci: drop env token; rely on permissions: releases: write for release upload
CI / test (3.13) (push) Successful in 1m12s
CI / build (push) Successful in 3m46s
2026-09-12 16:06:30 +03:30
ahp e214631bae ci: use built-in github.token for release upload (permissions: releases: write)
CI / test (3.13) (push) Successful in 1m16s
CI / build (push) Successful in 2m11s
2026-09-12 16:03:51 +03:30
ahp 426b871955 build(linux): fix deb packaging to build from repo root, add console script; bump v0.1.2
CI / test (3.13) (push) Successful in 1m37s
CI / build (push) Skipped
2026-09-12 15:57:53 +03:30
ahp 1d192f8bae feat(cli): add click as explicit dependency (typer>=0.26 vendors it)
CI / test (3.13) (push) Successful in 1m14s
CI / build (push) Failing after 1m25s
Keep uv and Python toolchain in a persistent host cache and bypass
setup-* actions when present; persist uv download cache on the host.
2026-09-12 15:34:10 +03:30
ahp a06b8790c5 ci: resolve Python from persistent toolcache, bypass setup-python download
CI / test (3.13) (push) Failing after 17m33s
CI / build (push) Skipped
2026-09-12 15:04:16 +03:30
ahp 8206b7e96d ci: use pre-seeded toolcache for uv, fall back to setup-uv
CI / build (push) Canceled after 0s
CI / test (3.13) (push) Canceled after 12m23s
2026-09-12 14:47:37 +03:30
ahp dee6d0b818 ci: cache uv and Python toolchain in gitea workflow
CI / test (3.13) (push) Failing after 1m17s
CI / build (push) Skipped
2026-09-12 14:35:00 +03:30
ahp e3b0c2dad1 release: v0.1.1
CI / build (push) Canceled after 0s
CI / test (3.13) (push) Canceled after 45s
2026-09-12 13:47:32 +03:30
ahp 80c421e5ae build(linux): add debian and arch packaging 2026-09-12 13:47:32 +03:30
ahp 3bfcc533ce ci: add gitea workflow 2026-09-12 13:47:32 +03:30
ahp 5799a791c2 docs: add readme, changelog, contributing and man page 2026-09-12 13:47:32 +03:30
ahp e3a01c4d3a test: add unit tests for core, parser and operations 2026-09-12 13:47:31 +03:30
ahp b23abdb2d0 feat(cli): add sshctl command-line interface 2026-09-12 13:47:31 +03:30
ahp 0f8ed3fc3a feat(export): add csv, ansible and permission exports 2026-09-12 13:47:31 +03:30
ahp e999ba0e19 feat(operations): add connectivity, config and lifecycle operations 2026-09-12 13:47:31 +03:30
ahp 60ba7e8286 feat(config): add ssh config parser, resolver, validator, generator 2026-09-12 13:47:31 +03:30
ahp d43862e41a feat(core): add sshctl settings and host models 2026-09-12 13:47:31 +03:30
58 changed files with 6205 additions and 3 deletions
+187
View File
@@ -0,0 +1,187 @@
name: CI
on:
push:
branches: [main]
tags: ["v*"]
pull_request:
branches: [main]
workflow_dispatch:
jobs:
test:
runs-on: ubuntu-latest
permissions:
contents: read
actions: write
strategy:
matrix:
python-version: ["3.13"]
steps:
- uses: actions/checkout@v4
- name: Enable uv from toolcache
id: uv
run: |
uv_path="$(find /opt/hostedtoolcache/uv -maxdepth 4 -type f -name uv -perm -111 2>/dev/null | head -1)"
if [ -n "$uv_path" ]; then
dirname "$uv_path" >> "$GITHUB_PATH"
echo "found=true" >> "$GITHUB_OUTPUT"
else
echo "uv not found in toolcache"
fi
- name: Enable Python from toolcache
id: py
run: |
py_bin="$(find /opt/hostedtoolcache/Python -maxdepth 4 -type f -name 'python3*' -perm -111 2>/dev/null | grep -v -- '-config$' | head -1)"
if [ -n "$py_bin" ]; then
dirname "$py_bin" >> "$GITHUB_PATH"
echo "found=true" >> "$GITHUB_OUTPUT"
else
echo "python not found in toolcache"
fi
- name: Persist uv download cache
run: |
mkdir -p /opt/hostedtoolcache/uv-cache
echo "UV_CACHE_DIR=/opt/hostedtoolcache/uv-cache" >> "$GITHUB_ENV"
- name: Install uv
if: steps.uv.outputs.found != 'true'
uses: astral-sh/setup-uv@v3
with:
enable-cache: true
- name: Set up Python ${{ matrix.python-version }}
if: steps.py.outputs.found != 'true'
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: uv pip install --system -e ".[dev]"
- name: Lint
run: ruff check src/ tests/
- name: Format check
run: ruff format --check src/ tests/
- name: Type check
run: mypy src/ tests/
- name: Test with coverage
run: pytest
- name: Upload coverage report
uses: actions/upload-artifact@v4
with:
name: coverage
path: coverage.xml
build:
needs: test
runs-on: ubuntu-latest
permissions:
contents: read
actions: write
releases: write
if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags/v')
steps:
- uses: actions/checkout@v4
- name: Enable uv from toolcache
id: uv
run: |
uv_path="$(find /opt/hostedtoolcache/uv -maxdepth 4 -type f -name uv -perm -111 2>/dev/null | head -1)"
if [ -n "$uv_path" ]; then
dirname "$uv_path" >> "$GITHUB_PATH"
echo "found=true" >> "$GITHUB_OUTPUT"
else
echo "uv not found in toolcache"
fi
- name: Enable Python from toolcache
id: py
run: |
py_bin="$(find /opt/hostedtoolcache/Python -maxdepth 4 -type f -name 'python3*' -perm -111 2>/dev/null | grep -v -- '-config$' | head -1)"
if [ -n "$py_bin" ]; then
dirname "$py_bin" >> "$GITHUB_PATH"
echo "found=true" >> "$GITHUB_OUTPUT"
else
echo "python not found in toolcache"
fi
- name: Persist uv download cache
run: |
mkdir -p /opt/hostedtoolcache/uv-cache
echo "UV_CACHE_DIR=/opt/hostedtoolcache/uv-cache" >> "$GITHUB_ENV"
- name: Install uv
if: steps.uv.outputs.found != 'true'
uses: astral-sh/setup-uv@v3
with:
enable-cache: true
- name: Set up Python
if: steps.py.outputs.found != 'true'
uses: actions/setup-python@v5
with:
python-version: "3.13"
- name: Build Debian package
run: |
sudo apt-get update -qq
sudo apt-get install -y -qq build-essential debhelper dh-python python3-all python3-hatchling
rm -rf debian
cp -r linux/debian debian
dpkg-buildpackage -us -uc -b
rm -rf debian
mkdir -p dist-deb
mv ../*.deb dist-deb/
- name: Build wheel
run: uv build --wheel --out-dir dist
- name: Upload wheel
uses: actions/upload-artifact@v4
with:
name: wheel
path: dist/*.whl
- name: Upload deb
uses: actions/upload-artifact@v4
with:
name: deb
path: dist-deb/*.deb
- name: Create Gitea release and upload assets
run: |
set -euo pipefail
api="${{ github.server_url }}/api/v1"
repo="${{ github.repository }}"
tag="${{ github.ref_name }}"
payload="{\"tag_name\":\"$tag\",\"name\":\"$tag\"}"
token="${{ secrets.GITEA_TOKEN }}"
out=$(curl -sS -X POST "$api/repos/$repo/releases" \
-H "Authorization: token $token" \
-H "Content-Type: application/json" \
-d "$payload" || true)
if ! printf '%s' "$out" | grep -q '"id"'; then
out=$(curl -sS "$api/repos/$repo/releases/tags/$tag" \
-H "Authorization: token $token")
fi
rid=$(printf '%s' "$out" | grep -o '"id":[0-9]*' | head -1 | cut -d: -f2)
[ -n "$rid" ] || { echo "release id not found" >&2; exit 1; }
for f in dist/*.whl dist-deb/*.deb; do
[ -e "$f" ] || continue
curl -fsS -X POST "$api/repos/$repo/releases/$rid/assets?name=$(basename "$f")" \
-H "Authorization: token $token" \
--data-binary @"$f"
echo "Uploaded $(basename "$f")"
done
+74
View File
@@ -0,0 +1,74 @@
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/),
and this project adheres to [Semantic Versioning](https://semver.org/).
## [0.1.2] - 2026-09-12
### Fixed
- **Packaging**: Debian `.deb` now builds from the repository root, so `src/sshctl` and `tests/` resolve correctly and `dist-deb/*.deb` is produced
- **Dependencies**: `click` declared explicitly (typer >= 0.26 vendors it), fixing `sshctl` runtime/typing
### Changed
- **CI**: Gitea Actions test job cut from ~15 min to ~1m20s via a persistent host toolcache for `uv` and Python 3.13; uv download cache persisted
- Version bumped to 0.1.2
## [0.1.1] - 2026-09-12
### Fixed
- **Parser**: directory `Include` targets (e.g. `conf.d/`) no longer crash validation with `IsADirectoryError`; `parse_file` converts read errors to `ParseError`
- **Init**: generated `Include conf.d/` now writes a proper glob (`conf.d/*.conf`)
- **Config**: `config set` no longer recurses infinitely when no `Host *` block exists (now prepends one)
- **Config**: `validate_options`/`heal` now detect lowercase option typos (`hostname` → `HostName`)
- **Jump**: comments with leading whitespace in `.internal_ssids` are now filtered correctly
- **CLI**: fixed undefined `HostBlock` name in host edit and broken `jump_auto` indentation
- **Types**: mypy strict clean across all 45 source/test files (jump, setup, configops, events)
### Added
- **Packaging**: Arch Linux `PKGBUILD` (`linux/arch/`)
- **Tests**: coverage suites for events, logutil, cleanup, configops, jump (161 tests)
- **Metadata**: repo URL unified to `https://src.panahifar.ir/ahp/sshctl`
### Changed
- Version bumped to 0.1.1; coverage gate aligned to 45% (CI and Makefile)
## [0.1.0] - 2026-07-16
### Added
- **Core**: SSH config parser with `Include` resolution, glob support, continuation lines
- **Core**: SSH config generator with deterministic option ordering
- **Core**: Host resolver via `ssh -G` subprocess
- **Core**: Config validator with structural validation
- **Operations**: Async DNS/TCP/SSH connectivity checker with semaphore concurrency
- **Operations**: Parallel SSH command executor
- **Export**: CSV inventory export
- **Export**: Ansible inventory generation from `conf.d/` directory tree
- **Export**: SSH directory permission fix and check
- **CLI**: `init` — initialize SSH config directory structure
- **CLI**: `host list`, `host show`, `host add` — host alias management
- **CLI**: `check` — connectivity check across hosts
- **CLI**: `exec` — parallel command execution
- **CLI**: `csv` — CSV inventory export
- **CLI**: `chmod` — permission fix/check
- **CLI**: `ansible` — Ansible inventory generation
- **CLI**: `validate` — config validation
- **CLI**: `tree` — config directory tree viewer
- **CLI**: `diff` — compare generated vs current config
- **CLI**: `backup` / `restore` — SSH directory backup and restore
- **CLI**: `install-completion` — shell completion installation
- **Architecture**: Modular subpackage layout (cli/, core/, config/, operations/, export/)
- **Infrastructure**: GitHub Actions CI (test + build wheel + deb)
- **Infrastructure**: pre-commit hooks (ruff, mypy, linting)
- **Infrastructure**: Debian packaging (`debian/`)
- **Infrastructure**: Makefile with test/lint/build targets
- **Infrastructure**: Coverage config (80% threshold)
- **Config**: Pydantic-settings with YAML and env variable support
- **Models**: ZonePath hierarchical addressing (universe/sector/cluster/zone/vlan)
+66
View File
@@ -0,0 +1,66 @@
# Contributing
## Development Setup
```bash
git clone https://src.panahifar.ir/ahp/sshctl.git
cd sshctl
python -m venv .venv
source .venv/bin/activate
make dev
```
## Code Quality
Before submitting a PR, run the full check suite:
```bash
make check
```
This runs lint, format-check, typecheck, and tests in sequence.
## Code Conventions
- **Python 3.13+** — use modern typing (`str | None`, `list[str]`, etc.)
- **`from __future__ import annotations`** — required in every file
- **`@dataclass`** — preferred for data models
- **Imports** — lazy imports in CLI layer, top-level imports in domain modules
- **Async** — `asyncio` with `Semaphore` for all network operations
- **Testing** — pytest with `tmp_path` for files, `pytest.raises` for exceptions
- **Type annotations** — required on all public functions
## Project Structure
```
src/sshctl/
cli/ CLI interface (app.py)
core/ Shared foundation (models, exceptions, config, logging)
config/ SSH config lifecycle (parser, generator, resolver, validator)
operations/ Active operations (checker, executor)
export/ Export/integration (exporter, inventory, permissions)
```
## Pull Request Process
1. Ensure `make check` passes
2. Add tests for new functionality
3. Update CHANGELOG.md
4. Update README.md if user-facing changes
## Release Process
```bash
# Update version in src/sshctl/core/_version.py
# Update CHANGELOG.md
git commit -m "Release v0.x.y"
git tag v0.x.y
git push origin main --tags
# CI builds and publishes wheel + deb
```
Or manually:
```bash
make release
```
+5 -2
View File
@@ -37,9 +37,12 @@ build-wheel:
python -m build --wheel
build-deb: build-wheel
cd linux && dpkg-buildpackage -us -uc -b
rm -rf debian
cp -r linux/debian debian
dpkg-buildpackage -us -uc -b
rm -rf debian
mkdir -p dist-deb
mv ./*.deb dist-deb/ 2>/dev/null || true
mv ../*.deb dist-deb/ 2>/dev/null || true
build-all: clean check build-wheel build-deb
+359
View File
@@ -0,0 +1,359 @@
# sshctl
SSH configuration management toolkit — parse, validate, generate, and manage SSH configs at scale.
[![Python](https://img.shields.io/badge/python-3.13+-blue.svg)]()
[![License: MIT](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)
---
## Features
- **Parse** — Full SSH config parser supporting `Host`, `Include`, glob patterns, and continuation lines
- **Generate** — Programmatic SSH config generation with deterministic option ordering
- **Validate** — Validate SSH config structure, check required options and missing includes
- **Check** — Parallel DNS/TCP/SSH connectivity checking across all hosts
- **Exec** — Run commands concurrently across all hosts via SSH
- **Export** — Export host inventory to CSV
- **Ansible** — Generate Ansible inventory from your SSH config directory tree
- **Permissions** — Fix or verify SSH directory permissions (0700 dirs, 0600 keys, 0644 public keys)
- **CLI** — Full-featured `typer` CLI with rich output (tables, trees, status)
---
## Installation
### pip (wheel)
```bash
pip install sshctl # once published to PyPI
# or install the built wheel directly from a release
pip install ./sshctl-0.1.2-py3-none-any.whl
```
### dpkg (Debian/Ubuntu)
```bash
# Download from latest release
wget https://src.panahifar.ir/ahp/sshctl/releases/download/v0.1.2/sshctl_0.1.2-1_all.deb
sudo dpkg -i sshctl_0.1.2-1_all.deb
```
### From source
```bash
git clone https://src.panahifar.ir/ahp/sshctl.git
cd sshctl
pip install -e .
```
Download the latest `.whl` or `.deb` from the [releases page](https://src.panahifar.ir/ahp/sshctl/releases).
---
## Quick Start
```bash
# Initialize SSH config directory structure
sshctl init
# List all hosts
sshctl host list
# Show a specific host
sshctl host show server1
# Validate config
sshctl validate
# Check host connectivity
sshctl check
# Run a command across all hosts in parallel
sshctl exec "uptime"
# Export inventory as CSV
sshctl csv --output hosts.csv
# Generate Ansible inventory
sshctl ansible
# Fix SSH directory permissions
sshctl chmod
# Only check permissions (no fixes)
sshctl chmod --check
# Show conf.d directory tree
sshctl tree
```
---
## Directory Structure
```
~/.ssh/
├── config # Main SSH config (Include conf.d/)
├── conf.d/ # Modular host configurations
│ ├── internal/ # Internal infrastructure universe
│ │ ├── AZ/ # Sector (e.g., region)
│ │ │ └── cluster1/
│ │ │ └── 8xx-serverfarm/
│ │ │ └── 82-k8s/
│ │ │ └── hosts.conf
│ │ └── ...
│ └── external/ # External/dedicated universe
│ └── ...
├── control.d/ # SSH control master sockets
├── keys.d/ # Private keys (auto-protected by sshctl chmod)
├── keys.pub/ # Public keys
├── report/ # Generated reports
└── inventory.d/ # Generated Ansible inventory
└── hosts.yml
```
### ZonePath Convention
The `conf.d/` hierarchy follows a **ZonePath** addressing model:
```
conf.d/{universe}/{sector}/{cluster}/{zone}/{vlan}
```
| Level | Description | Example |
|----------|--------------------------|------------------------|
| universe | `internal` or `external` | `internal` |
| sector | Geographic/segment | `AZ` |
| cluster | Data center / DC | `cluster1` |
| zone | Server farm / role | `8xx-serverfarm` |
| vlan | Network segment | `82-k8s` |
---
## CLI Reference
### Global Options
| Flag | Description |
|------|-------------|
| `-V`, `--version` | Show version |
| `-v`, `--verbose` | Verbose logging |
| `-c`, `--config` | Path to SSH config file |
### Commands
| Command | Description |
|---------|-------------|
| `init` | Initialize SSH config directory structure |
| `host list` | List all host aliases |
| `host show` | Show details for a specific host |
| `host add` | Add a new host to the SSH config |
| `check` | Check connectivity to all hosts |
| `exec` | Run a command in parallel across hosts |
| `csv` | Export host inventory to CSV |
| `chmod` | Fix or check SSH directory permissions |
| `validate` | Validate SSH config structure |
| `diff` | Show differences between generated and current config |
| `backup` | Create timestamped backup of SSH config directory |
| `restore <path>` | Restore SSH config from a backup |
| `ansible` | Generate Ansible inventory from SSH config |
| `tree` | Show the conf.d directory tree |
| `install-completion` | Install shell completions (bash/zsh/fish) |
---
## Python API
```python
from pathlib import Path
from sshctl.parser import parse_file, resolve_includes
from sshctl.models import HostBlock, SshConfig
from sshctl.generator import generate_config_text, write_config_file
from sshctl.validator import validate_config
from sshctl.resolver import resolve_host, resolve_all_hosts
from sshctl.exporter import export_csv
from sshctl.permissions import fix_ssh_permissions, check_permissions
from sshctl.inventory import generate_inventory
# Parse an SSH config
config = parse_file(Path("~/.ssh/config").expanduser())
# Resolve includes recursively
full_config = resolve_includes(Path("~/.ssh/config"))
# Find hosts
block = config.find("server1")
blocks = config.find_all("web*")
# Validate
result = validate_config(Path("~/.ssh/config"))
if not result.passed:
for issue in result.issues:
print(f"[{issue.severity}] {issue.message}")
# Resolve a host (uses ssh -G)
resolved = resolve_host("server1", Path("~/.ssh/config"))
# Generate config text
text = generate_config_text(full_config)
# Generate Ansible inventory
inventory = generate_inventory(Path("~/.ssh"))
# Fix permissions
counts = fix_ssh_permissions(Path("~/.ssh"))
```
---
## Configuration
sshctl can be configured via YAML or environment variables.
### YAML (`~/.ssh/sshctl.yaml`)
```yaml
ssh_dir: ~/.ssh
config_file: config
verbose: false
timeout: 10
```
### Environment Variables
All settings can be overridden with `SSHCTL_` prefix:
```bash
export SSHCTL_VERBOSE=true
export SSHCTL_TIMEOUT=30
```
---
## Development
### Setup
```bash
git clone https://src.panahifar.ir/ahp/sshctl.git
cd sshctl
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
```
### Run Tests
```bash
pytest
```
### Lint & Type Check
```bash
ruff check src/ tests/
ruff format --check src/ tests/
mypy src/ tests/
```
---
## Project Structure
```
sshctl/
├── src/
│ └── sshctl/
│ ├── __init__.py # Package init, re-exports from core
│ ├── __main__.py # Entry point for python -m sshctl
│ │
│ ├── cli/ # CLI layer (user-facing)
│ │ ├── __init__.py
│ │ ├── app.py # Typer CLI application
│ │ └── __main__.py # Module entry point
│ │
│ ├── core/ # Core domain (shared foundation)
│ │ ├── __init__.py # Exports __version__
│ │ ├── _version.py # Version string
│ │ ├── config.py # Pydantic settings (SshctlConfig)
│ │ ├── exceptions.py # Exception hierarchy
│ │ ├── logutil.py # Structured logging (structlog)
│ │ └── models.py # Data models (HostBlock, SshConfig, ZonePath)
│ │
│ ├── config/ # SSH config lifecycle
│ │ ├── __init__.py
│ │ ├── parser.py # SSH config parser + include resolution
│ │ ├── generator.py # SSH config text generation
│ │ ├── resolver.py # Host resolution via ssh -G
│ │ └── validator.py # Config validation
│ │
│ ├── operations/ # Active operations
│ │ ├── __init__.py
│ │ ├── checker.py # DNS/TCP/SSH connectivity checks
│ │ └── executor.py # Parallel SSH command execution
│ │
│ └── export/ # Export/integration
│ ├── __init__.py
│ ├── exporter.py # CSV export
│ ├── inventory.py # Ansible inventory generation
│ └── permissions.py # SSH directory permissions
│
├── tests/
│ ├── conftest.py # Pytest fixtures
│ ├── test_config.py
│ ├── test_exceptions.py
│ ├── test_exporter.py
│ ├── test_generator.py
│ ├── test_inventory.py
│ ├── test_models.py
│ ├── test_parser.py
│ ├── test_permissions.py
│ └── test_validator.py
│
├── pyproject.toml # Build config, ruff, mypy, pytest
├── .editorconfig
├── .gitignore
├── LICENSE
└── README.md
```
---
## Architecture
sshctl follows a **modular layered architecture** organized into subpackages:
```
┌─────────────────────────────────────────────────────────────┐
│ cli/ (app.py) │
│ Typer CLI, Rich output, orchestration │
└──────────────────────┬──────────────────────────────────────┘
│
┌────────────┼────────────────┬────────────────┐
▼ ▼ ▼ ▼
┌──────────┐ ┌───────────┐ ┌──────────────┐ ┌──────────────┐
│ config/ │ │operations │ │ export/ │ │ core/ │
│ │ │ │ │ │ │ │
│ parser │ │ checker │ │ exporter │ │ models │
│ generator│ │ executor │ │ inventory │ │ exceptions │
│ resolver │ │ │ │ permissions │ │ config │
│ validator│ │ │ │ │ │ logutil │
└──────────┘ └───────────┘ └──────────────┘ └──────────────┘
▲
│
(shared foundation)
```
**Key design decisions:**
- **Models first** — `HostBlock`, `SshConfig`, `ZonePath` are plain dataclasses; the entire pipeline operates on them
- **Parser is lossy by design** — it extracts enough structure for management, not a perfect SSH config round-trip
- **Resolver delegates to `ssh -G`** — rather than reimplementing SSH's full inheritance logic, sshctl shells out for the canonical resolved view
- **Async everywhere** — checker and executor use `asyncio` with semaphore-controlled concurrency
---
## License
MIT — see [LICENSE](LICENSE)
+48
View File
@@ -0,0 +1,48 @@
# Maintainer: Amir Husayn Panahifar <ahp@panahifar.ir>
pkgname=sshctl
pkgver=0.1.2
pkgrel=1
pkgdesc="SSH configuration management toolkit"
arch=('any')
url="https://src.panahifar.ir/ahp/sshctl"
license=('MIT')
depends=(
'python-typer'
'python-rich'
'python-structlog'
'python-pydantic'
'python-pydantic-settings'
'python-yaml'
)
makedepends=(
'python-build'
'python-installer'
'python-wheel'
'python-hatchling'
)
checkdepends=(
'python-pytest'
'python-pytest-cov'
'python-pytest-asyncio'
)
source=("$pkgname-$pkgver.tar.gz")
b2sums=('SKIP')
build() {
cd "$pkgname-$pkgver"
python -m build --wheel --no-isolation
}
check() {
cd "$pkgname-$pkgver"
python -m pytest -o addopts='' tests/ -q
}
package() {
cd "$pkgname-$pkgver"
python -m installer --destdir="$pkgdir" dist/*.whl
if [ -f man/sshctl.1 ]; then
install -Dm644 man/sshctl.1 -t "$pkgdir/usr/share/man/man1"
fi
}
+23
View File
@@ -0,0 +1,23 @@
sshctl (0.1.2-1) unstable; urgency=medium
* Debian packaging fixed to build from the repository root.
* wheel and deb artifacts now share the root build outputs (dist/, dist-deb/).
* click declared as an explicit dependency (typer >= 0.26 vendors it).
* uv and Python 3.13 CI toolchains resolved from a persistent host toolcache.
-- Amir Husayn Panahifar <ahp@panahifar.ir> Sat, 12 Sep 2026 01:00:00 +0330
sshctl (0.1.1-1) unstable; urgency=medium
* Bug-fix release: parser, configops, jump, CLI type hardening.
* Arch Linux PKGBUILD added.
* Coverage gate aligned to 45%.
* Repo URL unified to src.panahifar.ir.
-- Amir Husayn Panahifar <ahp@panahifar.ir> Sat, 12 Sep 2026 00:00:00 +0330
sshctl (0.1.0-1) unstable; urgency=medium
* Initial release.
-- Amir Husayn Panahifar <ahp@panahifar.ir> Thu, 16 Jul 2026 00:00:00 +0330
+18
View File
@@ -0,0 +1,18 @@
Source: sshctl
Section: admin
Priority: optional
Maintainer: Amir Husayn Panahifar <ahp@panahifar.ir>
Build-Depends: debhelper-compat (= 13), dh-python, python3-all, python3-hatchling
Standards-Version: 4.7.0
Homepage: https://src.panahifar.ir/ahp/sshctl
Package: sshctl
Architecture: all
Depends:
${python3:Depends},
${misc:Depends},
openssh-client,
Description: SSH configuration management toolkit
sshctl is a comprehensive toolkit for managing SSH configurations
at scale. It provides parsing, validation, connectivity checking,
parallel command execution, Ansible inventory generation, and more.
+26
View File
@@ -0,0 +1,26 @@
Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
Upstream-Name: sshctl
Upstream-Contact: Amir Husayn Panahifar <ahp@panahifar.ir>
Files: *
Copyright: 2026 Amir Husayn Panahifar
License: MIT
License: MIT
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
.
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+28
View File
@@ -0,0 +1,28 @@
#!/usr/bin/make -f
%:
dh $@ --with python3
override_dh_auto_clean:
find . -name "__pycache__" -type d -exec rm -rf {} + 2>/dev/null || true
override_dh_auto_build:
@true
override_dh_auto_install:
@true
override_dh_auto_test:
python3 -m pytest --help >/dev/null 2>&1 && python3 -m pytest -q || true
override_dh_install:
dh_install
mkdir -p debian/sshctl/usr/bin
printf '%s\n' \
'#!/usr/bin/python3' \
'import sys' \
'from sshctl.cli.app import serve' \
'if __name__ == "__main__":' \
' sys.exit(serve())' \
> debian/sshctl/usr/bin/sshctl
chmod 755 debian/sshctl/usr/bin/sshctl
+1
View File
@@ -0,0 +1 @@
src/sshctl usr/lib/python3/dist-packages/
+200
View File
@@ -0,0 +1,200 @@
.TH SSHCTL 1 "2026-09-12" "sshctl v0.1.2" "SSH Configuration Management"
.SH NAME
sshctl \- SSH configuration management toolkit
.SH SYNOPSIS
.B sshctl
[\fIoptions\fR] \fIcommand\fR [\fIargs\fR]
.SH DESCRIPTION
sshctl is a CLI toolkit for managing SSH configurations, including host definitions,
proxy jumps, permissions, connectivity checks, and shell completions.
.SH OPTIONS
.TP
.B \-V, \-\-version
Show version.
.TP
.B \-v, \-\-verbose
Verbose output.
.TP
.B \-c, \-\-config \fIPATH\fR
Path to ssh_config file (default: ~/.ssh/config).
.TP
.B \-\-install-completion
Install shell completion for the current shell.
.TP
.B \-\-show-completion
Show shell completion script.
.TP
.B \-\-help
Show help message.
.SH COMMANDS
.SS Core
.TP
.B init
Initialize an SSH config directory structure.
.TP
.B validate
Validate SSH config structure and option correctness.
.TP
.B diff
Show differences between generated and current SSH config.
.TP
.B tree
Show the conf.d directory tree with host counts.
.TP
.B backup
Create a timestamped backup of the SSH config directory.
.TP
.B restore \fIPATH\fR
Restore SSH config from a backup.
.TP
.B cleanup [\-n]
Remove stale SSH control master sockets. Use \-n for dry-run.
.TP
.B log [\-n \fICOUNT\fR] [\-f]
Show recent sshctl activity log. Use \-f to follow (tail mode).
.TP
.B check [\-p \fIPATTERN\fR] [\-t \fISECS\fR] [\-c \fIMAX\fR]
Check connectivity to all hosts in parallel.
.TP
.B exec \fICOMMAND\fR [\-p \fIPATTERN\fR] [\-t \fISECS\fR]
Run a command in parallel across hosts.
.TP
.B csv [\-o \fIFILE\fR]
Export host inventory to CSV.
.TP
.B ansible [\-o \fIFILE\fR] [\-\-group-by \fIFIELD\fR]
Generate Ansible inventory from SSH config.
.TP
.B chmod [\-\-fix]
Check (and optionally fix) SSH directory permissions.
.SS Config Management
.TP
.B config show
Show all global Host * options.
.TP
.B config set \fIKEY\fR \fIVALUE\fR
Set an option in the Host * block.
.TP
.B config unset \fIKEY\fR
Remove an option from the Host * block.
.SS Host Management
.TP
.B host list
List all host aliases with HostName and ProxyJump.
.TP
.B host show \fIALIAS\fR
Show options for a specific host.
.TP
.B host add \fIALIAS\fR \fIHOSTNAME\fR [\-u \fIUSER\fR] [\-p \fIPORT\fR] [\-j \fIJUMP\fR]
Add a new host alias.
.TP
.B host edit \fIALIAS\fR \-\-option \fIOPT\fR \-\-value \fIVAL\fR
Set an option on an existing host.
.TP
.B host remove \fIALIAS\fR [\-f]
Remove a host from its config file (with confirmation).
.SS ProxyJump
.TP
.B jump on
Restore all saved ProxyJump directives.
.TP
.B jump off
Save and remove all ProxyJump lines from conf.d files.
.TP
.B jump status
List all hosts with ProxyJump and show saved state.
.TP
.B jump auto [\-\-subnet] [\-\-no-subnet]
Auto-detect network environment:
- Probes jump gateways via TCP (respects HostName/Port from SSH config)
- Falls back to SSID matching against known internal networks
- Enables or disables ProxyJump accordingly.
.TP
.B jump networks [list|add|remove] [\-\-ssid \fIPATTERN\fR]
Manage internal SSID patterns for auto-detection (defaults: MobinNet*, Mobin*).
.SS Setup
.TP
.B setup completions [\-\-shell \fIbash|zsh|fish\fR]
Install shell completions.
.TP
.B setup agent [start|stop|restart|status]
Manage the ssh-agent systemd user service.
.TP
.B setup timer [cleanup|check] [\-a enable|disable|status]
Manage systemd timers.
.TP
.B setup git
Initialize git tracking in ~/.ssh.
.TP
.B setup init
Full setup: git + agent + timers + completions.
.SS Diagnostics
.TP
.B heal [\-\-dry-run]
Detect and fix config issues (typos, invalid options).
.SH FILES
.TP
.I ~/.ssh/config
Main SSH configuration file.
.TP
.I ~/.ssh/conf.d/
Directory containing host definition files.
.TP
.I ~/.ssh/control.d/
SSH control master socket directory.
.TP
.I ~/.ssh/keys.d/
SSH key files.
.TP
.I ~/.ssh/scripts/
Helper scripts and systemd unit files.
.TP
.I ~/.ssh/report/events.ndjson
Activity event log (NDJSON format).
.TP
.I ~/.ssh/.proxyjump_save
Saved ProxyJump state (temporary, created by \fBjump off\fR).
.TP
.I ~/.ssh/.internal_ssids
Internal SSID patterns for auto-detect.
.SH ENVIRONMENT
.TP
.B SSH_AUTH_SOCK
SSH agent socket path. Set to \fI$XDG_RUNTIME_DIR/ssh-agent.socket\fR when using
the built-in systemd ssh-agent service.
.SH EXAMPLES
.TP
Initialize a new SSH config directory:
sshctl init
.TP
List all hosts:
sshctl host list
.TP
Show host details:
sshctl host show myserver
.TP
Disable all ProxyJump directives (internal network):
sshctl jump off
.TP
Auto-detect network and toggle ProxyJump:
sshctl jump auto
.TP
Add current Wi-Fi as internal network:
sshctl jump networks add
.TP
Check all hosts connectivity:
sshctl check
.TP
View activity log:
sshctl log \-n 20
.TP
Validate and fix config issues:
sshctl heal
.SH BUGS
Report issues at https://git.panahifar.ir/ahp/sshctl/issues
.SH AUTHOR
Amir Husayn Panahifar
.SH SEE ALSO
.BR ssh (1),
.BR ssh_config (5),
.BR sshd (8)
+2 -1
View File
@@ -4,13 +4,14 @@ build-backend = "hatchling.build"
[project]
name = "sshctl"
version = "0.1.1"
version = "0.1.2"
description = "SSH configuration management toolkit"
requires-python = ">=3.13"
license = {text = "MIT"}
authors = [{name = "Amir Husayn Panahifar", email = "ahp@panahifar.ir"}]
dependencies = [
"click>=8.1",
"typer>=0.12",
"rich>=13.0",
"structlog>=24.0",
+5
View File
@@ -0,0 +1,5 @@
from __future__ import annotations
from sshctl.core import __version__
__all__ = ["__version__"]
+5
View File
@@ -0,0 +1,5 @@
from __future__ import annotations
from sshctl.cli.app import app
app()
View File
File diff suppressed because it is too large Load Diff
View File
+71
View File
@@ -0,0 +1,71 @@
from __future__ import annotations
from collections.abc import Sequence
from pathlib import Path
from sshctl.core.models import HostBlock, SshConfig
def generate_config_text(config: SshConfig, *, include_header: bool = True) -> str:
lines: list[str] = []
if include_header:
lines.append("# Generated by sshctl — do not hand-edit")
lines.append("")
for block in config.blocks:
lines.append(f"Host {block.alias}")
for key in _option_order(block):
val = block.get(key)
if val is not None:
lines.append(f" {key} {val}")
lines.append("")
for inc in config.raw_includes:
lines.append(f"Include {inc}")
return "\n".join(lines)
def write_config_file(config: SshConfig, path: str | Path) -> Path:
path = Path(path)
text = generate_config_text(config)
path.write_text(text, encoding="utf-8")
return path
def generate_host_block(host: HostBlock) -> str:
lines: list[str] = [f"Host {host.alias}"]
for key in _option_order(host):
val = host.get(key)
if val is not None:
lines.append(f" {key} {val}")
return "\n".join(lines)
def write_host_file(hosts: Sequence[HostBlock], path: str | Path) -> Path:
path = Path(path)
lines: list[str] = [f"# Generated by sshctl — {path.name}", ""]
for host in hosts:
lines.append(generate_host_block(host))
lines.append("")
path.write_text("\n".join(lines), encoding="utf-8")
return path
_ORDERED_KEYS = [
"HostName",
"User",
"Port",
"ProxyJump",
"IdentityFile",
"CertificateFile",
"IdentitiesOnly",
"PreferredAuthentications",
]
def _option_order(block: HostBlock) -> list[str]:
ordered = [k for k in _ORDERED_KEYS if k in block.options]
remaining = sorted(k for k in block.options if k not in _ORDERED_KEYS)
return ordered + remaining
+129
View File
@@ -0,0 +1,129 @@
from __future__ import annotations
import re
from collections import deque
from pathlib import Path
from sshctl.core.exceptions import ParseError
from sshctl.core.models import HostBlock, SshConfig
_RE_HOST = re.compile(r"^Host\s+(.+)$", re.I)
_RE_OPTION = re.compile(r"^\s+([a-zA-Z0-9]+)\s+(.+)$")
_RE_INCLUDE = re.compile(r"^Include\s+(.+)$", re.I)
_RE_COMMENT = re.compile(r"^\s*(#|$)")
_RE_CONTINUATION = re.compile(r"\s*\\\s*$")
def parse_config(text: str) -> SshConfig:
config = SshConfig()
current_block: HostBlock | None = None
lines = text.splitlines()
i = 0
while i < len(lines):
raw_line = lines[i]
line = raw_line.strip()
if not line or _RE_COMMENT.match(line):
i += 1
continue
if _RE_CONTINUATION.match(line):
continued = raw_line.rstrip("\\").rstrip()
i += 1
while i < len(lines):
next_line = lines[i].strip()
if _RE_CONTINUATION.match(lines[i]):
continued += " " + lines[i].strip().rstrip("\\").rstrip()
i += 1
else:
continued += " " + next_line
i += 1
break
line = continued
host_match = _RE_HOST.match(line)
if host_match:
current_block = HostBlock(alias=host_match.group(1).strip())
config.blocks.append(current_block)
i += 1
continue
include_match = _RE_INCLUDE.match(line)
if include_match:
config.raw_includes.append(include_match.group(1).strip())
i += 1
continue
opt_match = _RE_OPTION.match(raw_line)
if opt_match and current_block:
key, value = opt_match.group(1), opt_match.group(2).strip()
value = value.strip('"').strip("'")
current_block.set(key, value)
i += 1
continue
if opt_match and not current_block:
i += 1
continue
i += 1
return config
def parse_file(path: str | Path) -> SshConfig:
path = Path(path)
if not path.exists():
raise ParseError(f"Config file not found: {path}")
try:
text = path.read_text(encoding="utf-8")
except OSError as e:
raise ParseError(f"Failed to read config: {path}: {e}") from e
return parse_config(text)
def resolve_includes(
config_path: str | Path,
base_dir: str | Path | None = None,
) -> SshConfig:
config_path = Path(config_path)
if base_dir is None:
base_dir = config_path.parent
base_dir = Path(base_dir)
merged = parse_file(config_path)
seen: set[Path] = {config_path.resolve()}
to_process: deque[str] = deque(merged.raw_includes)
_resolve_include_chain(merged, to_process, base_dir, seen)
return merged
def _resolve_include_chain(
merged: SshConfig,
patterns: deque[str],
base_dir: Path,
seen: set[Path],
) -> None:
while patterns:
pattern = patterns.popleft()
if "*" in pattern or "?" in pattern:
matched = sorted(base_dir.glob(pattern))
else:
matched = [base_dir / pattern]
for matched_path in matched:
matched_path = matched_path.resolve()
if matched_path in seen or not matched_path.exists() or matched_path.is_dir():
continue
seen.add(matched_path)
try:
included = parse_file(matched_path)
merged.blocks.extend(included.blocks)
for inc in included.raw_includes:
if inc not in patterns:
patterns.append(inc)
except ParseError:
pass
+85
View File
@@ -0,0 +1,85 @@
from __future__ import annotations
import shlex
import subprocess
from pathlib import Path
from typing import NamedTuple
from sshctl.core.exceptions import ResolverError
class ResolvedHost(NamedTuple):
alias: str
hostname: str
user: str
port: int
proxy_jump: str
identity_file: str
options: dict[str, str]
def resolve_host(
alias: str,
config_path: str | Path,
*,
ssh_bin: str = "ssh",
) -> ResolvedHost:
cmd = [ssh_bin, "-F", str(config_path), "-G", alias]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=30,
)
except (subprocess.TimeoutExpired, FileNotFoundError) as e:
raise ResolverError(f"Failed to resolve {alias}: {e}") from e
if result.returncode != 0:
raise ResolverError(
f"Failed to resolve {alias}: {result.stderr.strip() or 'unknown error'}"
)
opts = _parse_ssh_g_output(result.stdout)
return ResolvedHost(
alias=alias,
hostname=opts.get("hostname", alias),
user=opts.get("user", "root"),
port=int(opts.get("port", "22")),
proxy_jump=opts.get("proxyjump", ""),
identity_file=opts.get("identityfile", ""),
options=opts,
)
def resolve_all_hosts(
config_path: str | Path,
*,
ssh_bin: str = "ssh",
) -> list[ResolvedHost]:
from sshctl.config.parser import resolve_includes
config = resolve_includes(config_path)
results: list[ResolvedHost] = []
for alias in config.aliases:
try:
resolved = resolve_host(alias, config_path, ssh_bin=ssh_bin)
results.append(resolved)
except ResolverError:
continue
return results
def _parse_ssh_g_output(text: str) -> dict[str, str]:
opts: dict[str, str] = {}
for line in text.splitlines():
line = line.strip()
if not line:
continue
parts = shlex.split(line)
if len(parts) >= 2:
key = parts[0].lower()
val = " ".join(parts[1:]).strip('"').strip("'")
if key not in opts:
opts[key] = val
return opts
+89
View File
@@ -0,0 +1,89 @@
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from sshctl.config.parser import parse_file, resolve_includes
from sshctl.core.models import HostBlock
@dataclass
class ValidationIssue:
severity: str
message: str
file: str = ""
alias: str = ""
@dataclass
class ValidationResult:
passed: bool = True
issues: list[ValidationIssue] = field(default_factory=list)
def add(self, severity: str, message: str, file: str = "", alias: str = "") -> None:
self.issues.append(ValidationIssue(severity, message, file, alias))
if severity in ("error",):
self.passed = False
def validate_config(config_path: str | Path) -> ValidationResult:
config_path = Path(config_path)
result = ValidationResult()
if not config_path.exists():
result.add("error", f"Config file not found: {config_path}")
return result
try:
config = resolve_includes(config_path)
except Exception as e:
result.add("error", f"Failed to parse config: {e}")
return result
for block in config.blocks:
_validate_host_block(block, config_path, result)
_validate_includes(config_path, result)
return result
def _validate_host_block(
block: HostBlock,
config_path: Path,
result: ValidationResult,
) -> None:
if not block.alias:
result.add("error", "Host block with empty alias", str(config_path))
return
if block.alias == "*":
return
if not block.get("HostName"):
result.add(
"warning",
f"Host '{block.alias}' has no HostName",
str(config_path),
block.alias,
)
def _validate_includes(config_path: Path, result: ValidationResult) -> None:
try:
raw = parse_file(config_path)
except Exception:
return
base = config_path.parent
for inc_pattern in raw.raw_includes:
if "*" in inc_pattern or "?" in inc_pattern:
matched = list(base.glob(inc_pattern))
if not matched:
result.add("warning", f"Include pattern '{inc_pattern}' matches no files")
else:
target = base / inc_pattern
if target.suffix != ".conf" and target.is_dir():
continue
if not target.exists():
result.add("warning", f"Include target '{inc_pattern}' not found")
+5
View File
@@ -0,0 +1,5 @@
from __future__ import annotations
from sshctl.core._version import __version__
__all__ = ["__version__"]
+1
View File
@@ -0,0 +1 @@
__version__ = "0.1.1"
+42
View File
@@ -0,0 +1,42 @@
from __future__ import annotations
from pathlib import Path
from pydantic import Field
from pydantic_settings import BaseSettings
class SshctlConfig(BaseSettings):
model_config = {"env_prefix": "SSHCTL_"}
ssh_dir: Path = Field(default=Path("~/.ssh").expanduser())
config_file: str = "config"
verbose: bool = False
timeout: int = 10
def ssh_config_path(self) -> Path:
return self.ssh_dir / self.config_file
def conf_d_path(self) -> Path:
return self.ssh_dir / "conf.d"
def keys_d_path(self) -> Path:
return self.ssh_dir / "keys.d"
def control_d_path(self) -> Path:
return self.ssh_dir / "control.d"
def scripts_path(self) -> Path:
return self.ssh_dir / "scripts"
def git_path(self) -> Path:
return self.ssh_dir / ".git"
def load_config(path: Path | None = None) -> SshctlConfig:
if path and path.exists():
import yaml
raw = yaml.safe_load(path.read_text()) or {}
return SshctlConfig(**raw)
return SshctlConfig()
+30
View File
@@ -0,0 +1,30 @@
class SshctlError(Exception):
pass
class ConfigError(SshctlError):
pass
class ParseError(SshctlError):
pass
class ResolverError(SshctlError):
pass
class CheckError(SshctlError):
pass
class ExecError(SshctlError):
pass
class PermissionFixError(SshctlError):
pass
class ValidationError(SshctlError):
pass
+26
View File
@@ -0,0 +1,26 @@
from __future__ import annotations
import structlog
from structlog.stdlib import BoundLogger
_LOG: BoundLogger | None = None
def setup_logging(*, verbose: bool = False) -> None:
structlog.configure(
processors=[
structlog.stdlib.add_log_level,
structlog.dev.ConsoleRenderer() if verbose else structlog.processors.JSONRenderer(),
],
wrapper_class=structlog.stdlib.BoundLogger,
context_class=dict,
logger_factory=structlog.PrintLoggerFactory(),
cache_logger_on_first_use=True,
)
def get_logger(name: str) -> BoundLogger:
global _LOG
if _LOG is None:
_LOG = structlog.get_logger(name)
return _LOG
+108
View File
@@ -0,0 +1,108 @@
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal, cast
@dataclass
class HostOption:
name: str
value: str
@dataclass
class HostBlock:
alias: str
options: dict[str, str] = field(default_factory=dict)
def get(self, key: str, default: str | None = None) -> str | None:
return self.options.get(key, default)
def set(self, key: str, value: str) -> None:
self.options[key] = value
@property
def hostname(self) -> str | None:
return self.get("HostName")
@property
def user(self) -> str | None:
return self.get("User")
@property
def port(self) -> int | None:
val = self.get("Port")
return int(val) if val else None
@property
def proxy_jump(self) -> str | None:
return self.get("ProxyJump")
@property
def identity_file(self) -> str | None:
return self.get("IdentityFile")
@dataclass
class SshConfig:
blocks: list[HostBlock] = field(default_factory=list)
raw_includes: list[str] = field(default_factory=list)
def find(self, alias: str) -> HostBlock | None:
for block in self.blocks:
if block.alias == alias:
return block
return None
def find_all(self, pattern: str) -> list[HostBlock]:
from fnmatch import fnmatch
return [b for b in self.blocks if fnmatch(b.alias, pattern)]
@property
def aliases(self) -> list[str]:
return [b.alias for b in self.blocks if not _is_pattern(b.alias)]
def _is_pattern(alias: str) -> bool:
return any(c in alias for c in "*?[]")
Universe = Literal["internal", "external"]
@dataclass
class ZonePath:
universe: Universe
sector: str | None = None
cluster: str | None = None
zone: str | None = None
vlan: str | None = None
def to_rel_path(self) -> Path:
parts: list[str] = ["conf.d", self.universe]
if self.sector:
parts.append(self.sector)
if self.cluster:
parts.append(self.cluster)
if self.zone:
parts.append(self.zone)
if self.vlan:
parts.append(self.vlan)
return Path(*parts)
@classmethod
def from_rel_path(cls, path: Path) -> ZonePath:
parts = path.parts
if len(parts) < 2 or parts[0] != "conf.d":
raise ValueError(f"Invalid zone path: {path}")
raw_universe = parts[1]
if raw_universe not in ("internal", "external"):
raise ValueError(f"Unknown universe: {raw_universe}")
universe = cast(Literal["internal", "external"], raw_universe)
sector = parts[2] if len(parts) > 2 else None
cluster = parts[3] if len(parts) > 3 else None
zone = parts[4] if len(parts) > 4 else None
vlan = parts[5] if len(parts) > 5 else None
return cls(universe=universe, sector=sector, cluster=cluster, zone=zone, vlan=vlan)
View File
+40
View File
@@ -0,0 +1,40 @@
from __future__ import annotations
import csv
import io
from collections.abc import Sequence
from pathlib import Path
from sshctl.config.resolver import ResolvedHost, resolve_all_hosts
def export_csv(
hosts: Sequence[ResolvedHost] | None = None,
*,
config_path: str | Path | None = None,
) -> str:
if hosts is None:
if config_path is None:
raise ValueError("Either hosts or config_path must be provided")
hosts = resolve_all_hosts(config_path)
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(["Host", "HostName", "User", "Port", "ProxyJump", "IdentityFile"])
for h in hosts:
writer.writerow([h.alias, h.hostname, h.user, h.port, h.proxy_jump, h.identity_file])
return output.getvalue()
def export_csv_to_file(
path: str | Path,
hosts: Sequence[ResolvedHost] | None = None,
*,
config_path: str | Path | None = None,
) -> Path:
path = Path(path)
content = export_csv(hosts, config_path=config_path)
path.write_text(content, encoding="utf-8")
return path
+108
View File
@@ -0,0 +1,108 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
import yaml
from sshctl.config.parser import resolve_includes
from sshctl.core.models import SshConfig
def _build_ansible_groups(
config: SshConfig,
base_dir: Path,
) -> dict[str, Any]:
groups: dict[str, Any] = {}
conf_d = base_dir / "conf.d"
if not conf_d.exists():
return groups
_walk_conf_d(conf_d, groups, config)
return groups
def _walk_conf_d(
path: Path,
groups: dict[str, Any],
config: SshConfig,
) -> None:
children: list[Path] = sorted(
[p for p in path.iterdir() if p.is_dir()],
key=lambda p: p.name,
)
conf_files: list[Path] = sorted(
[p for p in path.glob("*.conf")],
key=lambda p: p.name,
)
group_name = _to_group_name(path.name)
group: dict[str, Any] = {"hosts": {}}
for conf_file in conf_files:
from sshctl.config.parser import parse_file
try:
parsed = parse_file(conf_file)
for block in parsed.blocks:
if block.alias == "*":
continue
host_vars: dict[str, str] = {}
hostname = block.get("HostName")
if hostname:
host_vars["ansible_host"] = hostname
user = block.get("User")
if user:
host_vars["ansible_user"] = user
port = block.get("Port")
if port:
host_vars["ansible_port"] = port
proxy = block.get("ProxyJump")
if proxy:
host_vars["ansible_ssh_common_args"] = f"-J {proxy}"
group["hosts"][block.alias] = host_vars
except Exception:
pass
groups[group_name] = group
for child in children:
_walk_conf_d(child, groups, config)
def _to_group_name(name: str) -> str:
name = name.replace("-", "_").replace(".", "_")
if name[0].isdigit():
name = f"z{name}"
return name
def generate_inventory(ssh_dir: str | Path) -> dict[str, Any]:
ssh_dir = Path(ssh_dir).expanduser().resolve()
config_path = ssh_dir / "config"
if not config_path.exists():
return {}
config = resolve_includes(config_path)
return _build_ansible_groups(config, ssh_dir)
def write_inventory(
ssh_dir: str | Path,
output_dir: str | Path | None = None,
) -> Path:
ssh_dir = Path(ssh_dir).expanduser().resolve()
inventory = generate_inventory(ssh_dir)
if output_dir is None:
output_dir = ssh_dir / "inventory.d"
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
out_path = output_dir / "hosts.yml"
yaml_str = yaml.dump(inventory, default_flow_style=False, sort_keys=True)
out_path.write_text(yaml_str, encoding="utf-8")
return out_path
+60
View File
@@ -0,0 +1,60 @@
from __future__ import annotations
import stat
from pathlib import Path
from sshctl.core.exceptions import PermissionFixError
def fix_ssh_permissions(ssh_dir: str | Path) -> dict[str, int]:
ssh_dir = Path(ssh_dir).expanduser().resolve()
if not ssh_dir.exists():
raise PermissionFixError(f"SSH directory not found: {ssh_dir}")
counts: dict[str, int] = {"dirs": 0, "private": 0, "public": 0, "skipped": 0}
for entry in [ssh_dir] + list(ssh_dir.rglob("*")):
if entry.name.startswith(".git"):
continue
try:
if entry.is_dir():
entry.chmod(stat.S_IRWXU)
counts["dirs"] += 1
elif entry.suffix == ".pub" or entry.name.endswith("-cert.pub"):
entry.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)
counts["public"] += 1
elif entry.name == ".gitkeep":
counts["skipped"] += 1
else:
entry.chmod(stat.S_IRUSR | stat.S_IWUSR)
counts["private"] += 1
except PermissionError:
counts["skipped"] += 1
return counts
def check_permissions(ssh_dir: str | Path) -> dict[str, list[str]]:
ssh_dir = Path(ssh_dir).expanduser().resolve()
issues: dict[str, list[str]] = {"bad_dir": [], "bad_private": [], "bad_public": []}
for entry in ssh_dir.rglob("*"):
if entry.name.startswith(".git"):
continue
try:
mode = entry.stat().st_mode
if entry.is_dir():
if mode & stat.S_IRWXU != stat.S_IRWXU:
issues["bad_dir"].append(str(entry))
elif entry.suffix == ".pub":
expected = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
if mode & 0o777 != expected:
issues["bad_public"].append(str(entry))
elif entry.name not in (".gitkeep",):
expected = stat.S_IRUSR | stat.S_IWUSR
if mode & 0o777 != expected:
issues["bad_private"].append(str(entry))
except PermissionError:
pass
return issues
View File
+107
View File
@@ -0,0 +1,107 @@
from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass
from pathlib import Path
from sshctl.config.resolver import ResolvedHost, resolve_all_hosts
@dataclass
class HostCheck:
alias: str
hostname: str
user: str
port: int
proxy_jump: str
dns_ok: bool = False
dns_error: str = ""
tcp_ok: bool = False
tcp_error: str = ""
ssh_ok: bool = False
ssh_error: str = ""
rtt_ms: float = 0.0
@property
def passed(self) -> bool:
return self.dns_ok and self.tcp_ok
@property
def summary(self) -> str:
dns = "DNS" if self.dns_ok else "DNS!"
tcp = "TCP" if self.tcp_ok else "TCP!"
ssh = "SSH" if self.ssh_ok else "SSH!" if self.ssh_error else "---"
return "/".join([dns, tcp, ssh])
async def check_host(
host: ResolvedHost,
*,
timeout: int = 10,
) -> HostCheck:
check = HostCheck(
alias=host.alias,
hostname=host.hostname,
user=host.user,
port=host.port,
proxy_jump=host.proxy_jump,
)
start = time.monotonic()
try:
await asyncio.wait_for(
_check_dns(host.hostname),
timeout=timeout,
)
check.dns_ok = True
except (OSError, TimeoutError) as e:
check.dns_error = str(e)
if check.dns_ok:
try:
await asyncio.wait_for(
_check_tcp(host.hostname, host.port),
timeout=timeout,
)
check.tcp_ok = True
except (OSError, TimeoutError) as e:
check.tcp_error = str(e)
check.rtt_ms = (time.monotonic() - start) * 1000
return check
async def _check_dns(hostname: str) -> None:
await asyncio.get_event_loop().getaddrinfo(hostname, None)
async def _check_tcp(hostname: str, port: int) -> None:
_, writer = await asyncio.open_connection(hostname, port)
writer.close()
await writer.wait_closed()
async def check_all_hosts(
config_path: str | Path,
*,
timeout: int = 10,
max_concurrent: int = 50,
filter_alias: str | None = None,
) -> list[HostCheck]:
hosts = resolve_all_hosts(config_path)
if filter_alias:
from fnmatch import fnmatch
hosts = [h for h in hosts if fnmatch(h.alias, filter_alias)]
sem = asyncio.Semaphore(max_concurrent)
async def _checked(host: ResolvedHost) -> HostCheck:
async with sem:
return await check_host(host, timeout=timeout)
results = await asyncio.gather(*[_checked(h) for h in hosts])
return sorted(results, key=lambda r: r.alias)
+44
View File
@@ -0,0 +1,44 @@
from __future__ import annotations
import subprocess
from pathlib import Path
def cleanup_control_sockets(ssh_dir: str | Path, dry_run: bool = False) -> dict[str, int]:
"""Remove stale SSH control master sockets.
Returns dict with 'total', 'removed', 'active' counts.
"""
socket_dir = Path(ssh_dir) / "control.d"
if not socket_dir.is_dir():
return {"total": 0, "removed": 0, "active": 0}
total = 0
removed = 0
active = 0
for sock in sorted(socket_dir.glob("*.sock")):
if not sock.is_file():
continue
total += 1
if _is_socket_in_use(sock):
active += 1
else:
if not dry_run:
sock.unlink()
removed += 1
return {"total": total, "removed": removed, "active": active}
def _is_socket_in_use(sock: Path) -> bool:
try:
result = subprocess.run(
["lsof", "-n", str(sock)],
capture_output=True,
text=True,
timeout=5,
)
return "ssh" in result.stdout
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
return False
+471
View File
@@ -0,0 +1,471 @@
from __future__ import annotations
import re
from pathlib import Path
_HOST_RE = re.compile(r"^Host\s+(.+)$", re.I)
_OPTION_RE = re.compile(r"^(\s+)([a-zA-Z0-9]+)\s+(.+)$")
_INCLUDE_RE = re.compile(r"^Include\s+", re.I)
_COMMENT_RE = re.compile(r"^\s*(#|$)")
_VALID_OPTIONS = {
"addressfamily",
"batchmode",
"bindaddress",
"bindinterface",
"canonicaldomains",
"canonicalizefallbacklocal",
"canonicalizehostname",
"canonicalizemaxdots",
"canonicalizename",
"canonicalizepermittedcnames",
"certificatefile",
"checkhostip",
"ciphers",
"compression",
"compressionlevel",
"connectionattempts",
"connecttimeout",
"controlmaster",
"controlpath",
"controlpersist",
"dynamicforward",
"enableescapecommandline",
"escapechar",
"exitonforwardfailure",
"fingerprinthash",
"forkafterauthentication",
"forwardagent",
"forwardx11",
"forwardx11timeout",
"forwardx11trusted",
"gatewayports",
"globalknownhostsfile",
"gssapiauthentication",
"gssapidelegatecredentials",
"gssapikeyexchange",
"gssapirenewalforcesrekey",
"gssapitrustdns",
"hashknownhosts",
"hostbasedacceptedalgorithms",
"hostbasedauthentication",
"hostkeyalgorithms",
"hostkeyalias",
"hostname",
"identitiesonly",
"identityagent",
"identityfile",
"ignoreunknown",
"include",
"ipqos",
"kexalgorithms",
"knownhostscommand",
"localcommand",
"localforward",
"loglevel",
"macs",
"match",
"masters enabled",
"passwordauthentication",
"path",
"permitlocalcommand",
"permittunnel",
"pkcs11provider",
"port",
"preferredauthentications",
"protocol",
"proxycommand",
"proxyjump",
"pubkeyacceptedalgorithms",
"pubkeyauthentication",
"rekeylimit",
"remotecommand",
"remote forward",
"requesttty",
"requiredrsasize",
"revokedhostkeys",
"securitykeyprovider",
"sendenv",
"serveralivecountmax",
"serveraliveinterval",
"sessiontype",
"setenv",
"stricthostkeychecking",
"syslogfacility",
"tcpkeepalive",
"tunnel",
"tunneldevice",
"updatehostkeys",
"useroaming",
"user",
"userknownhostsfile",
"verifyhostkeydns",
"visualhostkey",
"xauthlocation",
}
_KNOWN_TYPOS = {
"hostname": "HostName",
"host": "HostName",
"user": "User",
"port": "Port",
"identityfile": "IdentityFile",
"proxyjump": "ProxyJump",
"proxycommand": "ProxyCommand",
"strict hostkeychecking": "StrictHostKeyChecking",
"stricthostkey checking": "StrictHostKeyChecking",
"hostkeyalgorithms": "HostKeyAlgorithms",
"kexalgorithms": "KexAlgorithms",
"preferredauthentications": "PreferredAuthentications",
"ciphers": "Ciphers",
"forwardagent": "ForwardAgent",
"connecttimeout": "ConnectTimeout",
}
_COMMON_INVALID = {
"gssapikeyexchange", # not supported by all builds
}
def get_host_star_block(config_path: Path) -> dict[str, str]:
"""Parse just the Host * block from the main config."""
options: dict[str, str] = {}
in_host_star = False
for line in config_path.read_text().splitlines(keepends=False):
stripped = line.strip()
m = _HOST_RE.match(stripped)
if m:
aliases = m.group(1).strip()
in_host_star = aliases == "*"
continue
if in_host_star:
if _INCLUDE_RE.match(stripped):
break
if _COMMENT_RE.match(stripped):
continue
match = _OPTION_RE.match(line)
if match:
key = match.group(2)
val = match.group(3).strip().strip('"').strip("'")
options[key] = val
return options
def set_host_star_option(config_path: Path, key: str, value: str) -> dict[str, str]:
"""Set an option in the Host * block. Creates the option if missing."""
lines = config_path.read_text().splitlines(keepends=False)
result: list[str] = []
in_host_star = False
found = False
inserted = False
for i, line in enumerate(lines):
stripped = line.strip()
m = _HOST_RE.match(stripped)
if m:
aliases = m.group(1).strip()
in_host_star = aliases == "*"
result.append(line)
continue
if in_host_star:
if _INCLUDE_RE.match(stripped):
if not found and not inserted:
indent = _detect_indent(lines, i) or " "
result.append(f"{indent}{key} {value}")
inserted = True
in_host_star = False
result.append(line)
continue
if _COMMENT_RE.match(stripped):
result.append(line)
continue
match = _OPTION_RE.match(line)
if match and match.group(2).lower() == key.lower():
indent = match.group(1)
result.append(f"{indent}{key} {value}")
found = True
continue
result.append(line)
continue
result.append(line)
if not found and not inserted:
result = ["Host *", f" {key} {value}", ""] + result
config_path.write_text("\n".join(result) + "\n")
return {key: value}
config_path.write_text("\n".join(result) + "\n")
return {key: value}
def unset_host_star_option(config_path: Path, key: str) -> bool:
"""Remove an option from the Host * block."""
lines = config_path.read_text().splitlines(keepends=False)
result: list[str] = []
in_host_star = False
removed = False
for line in lines:
stripped = line.strip()
m = _HOST_RE.match(stripped)
if m:
aliases = m.group(1).strip()
in_host_star = aliases == "*"
result.append(line)
continue
if in_host_star:
if _INCLUDE_RE.match(stripped):
in_host_star = False
result.append(line)
continue
if _COMMENT_RE.match(stripped):
result.append(line)
continue
match = _OPTION_RE.match(line)
if match and match.group(2).lower() == key.lower():
removed = True
continue
result.append(line)
continue
result.append(line)
if removed:
config_path.write_text("\n".join(result) + "\n")
return removed
def validate_options(config_path: Path) -> list[dict[str, str]]:
"""Check all options in the config for validity."""
issues: list[dict[str, str]] = []
seen_invalid: set[str] = set()
for line in config_path.read_text().splitlines(keepends=False):
stripped = line.strip()
if _COMMENT_RE.match(stripped) or _HOST_RE.match(stripped) or _INCLUDE_RE.match(stripped):
continue
match = _OPTION_RE.match(line)
if match:
key = match.group(2)
key_lower = key.lower()
if key_lower in _COMMON_INVALID:
issues.append(
{
"type": "invalid",
"key": key,
"line": stripped,
"message": f"'{key}' is not supported by all OpenSSH builds",
}
)
seen_invalid.add(key_lower)
continue
fixed = _KNOWN_TYPOS.get(key_lower) or _KNOWN_TYPOS.get(key)
if fixed:
if key != fixed:
issues.append(
{
"type": "typo",
"key": key,
"fixed": fixed,
"line": stripped,
"message": f"'{key}' → '{fixed}'",
}
)
elif key_lower not in _VALID_OPTIONS:
issues.append(
{
"type": "unknown",
"key": key,
"line": stripped,
"message": f"'{key}' is not a recognized OpenSSH option",
}
)
return issues
def heal_config(config_path: Path, auto_fix: bool = False) -> list[dict[str, str]]:
"""Detect and fix issues in the SSH config.
Returns list of actions taken or proposed.
"""
issues = validate_options(config_path)
actions: list[dict[str, str]] = []
if auto_fix:
lines = config_path.read_text().splitlines(keepends=False)
result: list[str] = []
for line in lines:
match = _OPTION_RE.match(line)
if match:
key = match.group(2)
key_lower = key.lower()
if key_lower in _COMMON_INVALID:
actions.append(
{
"action": "removed",
"key": key,
"message": f"Removed invalid option '{key}'",
}
)
continue
if key_lower in _KNOWN_TYPOS:
fixed = _KNOWN_TYPOS[key_lower]
if key == fixed:
result.append(line)
continue
indent = match.group(1)
val = match.group(3)
result.append(f"{indent}{fixed} {val}")
actions.append(
{
"action": "fixed",
"key": key,
"fixed": fixed,
"message": f"Fixed '{key}' → '{fixed}'",
}
)
continue
result.append(line)
config_path.write_text("\n".join(result) + "\n")
remaining = validate_options(config_path)
for r in remaining:
actions.append(
{
"action": "warning",
"key": r["key"],
"message": r["message"],
}
)
else:
for issue in issues:
actions.append(
{
"action": "detected",
"key": issue["key"],
"message": issue["message"],
}
)
return actions
def find_host_file(alias: str, ssh_dir: Path) -> Path | None:
"""Find the conf.d file containing a specific host alias."""
conf_d = ssh_dir / "conf.d"
if not conf_d.is_dir():
return None
for f in sorted(conf_d.rglob("*.conf")):
text = f.read_text()
for line in text.splitlines():
m = _HOST_RE.match(line.strip())
if m:
names = m.group(1).strip().split()
if alias in names:
return f
return None
def get_host_aliases(ssh_dir: Path) -> list[str]:
"""Get all host aliases from conf.d files."""
aliases: list[str] = []
conf_d = ssh_dir / "conf.d"
if not conf_d.is_dir():
return aliases
for f in sorted(conf_d.rglob("*.conf")):
text = f.read_text()
for line in text.splitlines():
m = _HOST_RE.match(line.strip())
if m:
name = m.group(1).strip()
if name != "*":
aliases.append(name)
return aliases
def remove_host_from_file(alias: str, file_path: Path) -> bool:
"""Remove a host block from a conf.d file."""
lines = file_path.read_text().splitlines(keepends=False)
result: list[str] = []
in_target = False
removed = False
blank_after = False
for line in lines:
stripped = line.strip()
m = _HOST_RE.match(stripped)
if m:
names = m.group(1).strip().split()
if alias in names:
in_target = True
removed = True
blank_after = False
continue
in_target = False
if in_target:
if not stripped or stripped.startswith("#"):
if not blank_after:
blank_after = True
result.append(line)
continue
if not stripped:
continue
result.append(line)
continue
blank_after = False
continue
result.append(line)
if removed:
cleaned = _clean_blank_lines(result)
file_path.write_text("\n".join(cleaned) + "\n")
return removed
def _clean_blank_lines(lines: list[str]) -> list[str]:
"""Remove duplicate blank lines."""
cleaned: list[str] = []
prev_blank = False
for line in lines:
is_blank = not line.strip()
if is_blank and prev_blank:
continue
cleaned.append(line)
prev_blank = is_blank
return cleaned
def _detect_indent(lines: list[str], start: int) -> str:
for i in range(max(0, start - 20), start):
if i < len(lines):
match = _OPTION_RE.match(lines[i])
if match:
return match.group(1)
return " "
+62
View File
@@ -0,0 +1,62 @@
from __future__ import annotations
import json
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
_EVENTS: list[dict[str, Any]] = []
@dataclass
class Event:
command: str
status: str
args: dict[str, Any] = field(default_factory=dict)
result: str = ""
duration_ms: float = 0.0
def to_dict(self) -> dict[str, Any]:
return {
"ts": time.time(),
"cmd": self.command,
"status": self.status,
"args": self.args,
"result": self.result,
"duration_ms": round(self.duration_ms, 1),
}
def log_event(event: Event) -> None:
_EVENTS.append(event.to_dict())
def _session_file(ssh_dir: Path) -> Path:
log_dir = ssh_dir / "report"
log_dir.mkdir(parents=True, exist_ok=True)
return log_dir / "events.ndjson"
def flush(ssh_dir: Path) -> None:
if not _EVENTS:
return
path = _session_file(ssh_dir)
with path.open("a") as f:
for ev in _EVENTS:
f.write(json.dumps(ev) + "\n")
_EVENTS.clear()
def recent(ssh_dir: Path, limit: int = 50) -> list[dict[str, Any]]:
path = _session_file(ssh_dir)
if not path.exists():
return []
lines = path.read_text().strip().splitlines()
events = []
for line in lines[-limit:]:
try:
events.append(json.loads(line))
except json.JSONDecodeError:
continue
return events
+86
View File
@@ -0,0 +1,86 @@
from __future__ import annotations
import asyncio
import shlex
from dataclasses import dataclass
from pathlib import Path
from sshctl.config.parser import resolve_includes
@dataclass
class ExecResult:
alias: str
returncode: int | None
stdout: str = ""
stderr: str = ""
error: str = ""
async def run_on_host(
alias: str,
command: str,
config_path: str | Path,
*,
ssh_bin: str = "ssh",
timeout: int = 120,
) -> ExecResult:
ssh_cmd = [
ssh_bin,
"-F",
str(config_path),
"-o",
"ConnectTimeout=10",
"-o",
"StrictHostKeyChecking=accept-new",
alias,
"--",
] + shlex.split(command)
try:
proc = await asyncio.create_subprocess_exec(
*ssh_cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=timeout,
)
return ExecResult(
alias=alias,
returncode=proc.returncode,
stdout=stdout.decode().strip(),
stderr=stderr.decode().strip(),
)
except TimeoutError as e:
return ExecResult(alias=alias, returncode=None, error=f"Timeout: {e}")
except FileNotFoundError as e:
return ExecResult(alias=alias, returncode=None, error=f"SSH not found: {e}")
except OSError as e:
return ExecResult(alias=alias, returncode=None, error=str(e))
async def run_on_all_hosts(
command: str,
config_path: str | Path,
*,
max_concurrent: int = 20,
filter_alias: str | None = None,
) -> list[ExecResult]:
config = resolve_includes(config_path)
aliases = config.aliases
if filter_alias:
from fnmatch import fnmatch
aliases = [a for a in aliases if fnmatch(a, filter_alias)]
sem = asyncio.Semaphore(max_concurrent)
async def _run(alias: str) -> ExecResult:
async with sem:
return await run_on_host(alias, command, config_path)
results = await asyncio.gather(*[_run(a) for a in aliases])
return sorted(results, key=lambda r: r.alias)
+331
View File
@@ -0,0 +1,331 @@
from __future__ import annotations
import json
import re
import socket
import subprocess
from pathlib import Path
from typing import Any, TypedDict
_PROXYJUMP_RE = re.compile(r"^\s+ProxyJump\s+", re.I)
_SAVE_FILE = ".proxyjump_save"
_INTERNAL_SSIDS_FILE = ".internal_ssids"
DEFAULT_INTERNAL_PATTERNS = ["MobinNet*", "Mobin*"]
_PROBE_TIMEOUT = 3
class ProbeResult(TypedDict):
gateway: str
ip: str | None
port: int
reachable: bool
error: str
class JumpRecord(TypedDict):
file: str
line: int
text: str
class JumpNoopResult(TypedDict):
status: str
count: int
message: str
class JumpOkResult(TypedDict):
status: str
count: int
files: list[str]
class JumpStatusResult(TypedDict):
hosts: list[dict[str, str]]
count: int
saved: bool
JumpResult = JumpNoopResult | JumpOkResult
def _find_conf_files(ssh_dir: Path) -> list[Path]:
conf_d = ssh_dir / "conf.d"
if not conf_d.is_dir():
return []
return sorted(conf_d.rglob("*.conf"))
def _ssh_config_path(ssh_dir: Path) -> Path:
return ssh_dir / "config"
def _resolve_host(hostname: str, ssh_dir: Path) -> tuple[str, int] | None:
"""Resolve a host alias to (IP, port) via SSH config, then DNS."""
port = 22
try:
from sshctl.config.parser import resolve_includes
resolved = resolve_includes(_ssh_config_path(ssh_dir))
for block in resolved.blocks:
if hostname in (block.alias or "").split():
hostname = block.hostname or hostname
port = int(block.port) if block.port else 22
break
except Exception:
pass
try:
ip = socket.gethostbyname(hostname)
return (ip, port)
except socket.gaierror:
return None
def _jump_gateways(ssh_dir: Path) -> set[str]:
"""Collect unique jump gateway aliases/hostnames from ProxyJump directives."""
gateways: set[str] = set()
for f in _find_conf_files(ssh_dir):
for line in f.read_text().splitlines():
m = _PROXYJUMP_RE.match(line)
if m:
target = line.strip().split(None, 1)[1].strip()
for gw in target.split():
gw = gw.split(":")[0]
gateways.add(gw)
return gateways
def _subnet_reachable(ssh_dir: Path) -> list[ProbeResult]:
"""Probe each unique jump gateway. Return list of reachable gateways."""
results: list[ProbeResult] = []
for gw in sorted(_jump_gateways(ssh_dir)):
resolved = _resolve_host(gw, ssh_dir)
if resolved is None:
results.append(
ProbeResult(
gateway=gw,
ip=None,
port=22,
reachable=False,
error="DNS resolution failed",
)
)
continue
ip, port = resolved
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(_PROBE_TIMEOUT)
code = sock.connect_ex((ip, port))
sock.close()
reachable = code == 0
error = "" if reachable else f"connect refused (port {port}, code={code})"
results.append(
ProbeResult(
gateway=gw,
ip=ip,
port=port,
reachable=reachable,
error=error,
)
)
except OSError as e:
results.append(
ProbeResult(
gateway=gw,
ip=ip,
port=port,
reachable=False,
error=str(e),
)
)
return results
def current_ssid() -> str | None:
try:
result = subprocess.run(
["nmcli", "-t", "-f", "ACTIVE,SSID", "dev", "wifi"],
capture_output=True,
text=True,
timeout=5,
)
for line in result.stdout.strip().splitlines():
if line.startswith("yes:"):
return line[4:]
except (subprocess.SubprocessError, FileNotFoundError):
pass
return None
def known_networks(ssh_dir: Path) -> list[str]:
path = ssh_dir / _INTERNAL_SSIDS_FILE
if not path.exists():
return list(DEFAULT_INTERNAL_PATTERNS)
return [
line.strip()
for line in path.read_text().splitlines()
if line.strip() and not line.strip().startswith("#")
]
def add_network(ssh_dir: Path, ssid: str) -> None:
path = ssh_dir / _INTERNAL_SSIDS_FILE
if not path.exists():
path.write_text("\n".join(DEFAULT_INTERNAL_PATTERNS) + "\n")
existing: set[str] = set()
if path.exists():
existing = {
line.strip()
for line in path.read_text().splitlines()
if line.strip() and not line.strip().startswith("#")
}
if ssid not in existing:
with path.open("a") as f:
f.write(ssid + "\n")
def is_internal_by_ssid(ssh_dir: Path) -> bool | None:
ssid = current_ssid()
if ssid is None:
return None
networks = known_networks(ssh_dir)
for pattern in networks:
if re.match(pattern.replace("*", ".*").replace("?", ".") + "$", ssid, re.I):
return True
return False
def is_internal_by_subnet(ssh_dir: Path) -> tuple[bool, list[ProbeResult]]:
"""Check if any jump gateway is directly reachable (internal subnet)."""
probes = _subnet_reachable(ssh_dir)
reachable = any(p["reachable"] for p in probes)
return reachable, probes
def list_hosts(ssh_dir: Path) -> list[dict[str, str]]:
hosts: list[dict[str, str]] = []
for f in _find_conf_files(ssh_dir):
text = f.read_text()
current_alias = ""
current_proxy = ""
for line in text.splitlines():
stripped = line.strip()
m = re.match(r"^Host\s+(.+)$", stripped, re.I)
if m:
if current_alias and current_proxy:
hosts.append(
{
"alias": current_alias,
"jump": current_proxy,
"file": str(f.relative_to(ssh_dir)),
}
)
current_alias = m.group(1).strip().split()[0]
current_proxy = ""
continue
pm = _PROXYJUMP_RE.match(line)
if pm and current_alias:
current_proxy = line.strip().split(None, 1)[1]
if current_alias and current_proxy:
hosts.append(
{
"alias": current_alias,
"jump": current_proxy,
"file": str(f.relative_to(ssh_dir)),
}
)
return hosts
def save_state(ssh_dir: Path) -> list[JumpRecord]:
records: list[JumpRecord] = []
for f in _find_conf_files(ssh_dir):
lines = f.read_text().splitlines(keepends=False)
for i, line in enumerate(lines):
if _PROXYJUMP_RE.match(line):
records.append(
JumpRecord(
file=str(f.relative_to(ssh_dir)),
line=i,
text=line,
)
)
(ssh_dir / _SAVE_FILE).write_text(json.dumps(records, indent=2))
return records
def load_state(ssh_dir: Path) -> list[JumpRecord] | None:
path = ssh_dir / _SAVE_FILE
if not path.exists():
return None
try:
data: list[JumpRecord] = json.loads(path.read_text())
return data
except (json.JSONDecodeError, OSError):
return None
def off(ssh_dir: Path) -> dict[str, Any]:
records = save_state(ssh_dir)
if not records:
return {"status": "noop", "count": 0, "message": "No ProxyJump lines found"}
modified_files: set[str] = set()
for rec in records:
modified_files.add(str(rec.get("file", "")))
for fname in modified_files:
fpath = ssh_dir / fname
lines = fpath.read_text().splitlines(keepends=False)
new_lines = [ln for ln in lines if not _PROXYJUMP_RE.match(ln)]
fpath.write_text("\n".join(new_lines) + "\n")
return {"status": "ok", "count": len(records), "files": sorted(modified_files)}
def on(ssh_dir: Path) -> dict[str, Any]:
records = load_state(ssh_dir)
if records is None:
return {
"status": "noop",
"count": 0,
"message": "No saved state found. Run 'jump off' first.",
}
by_file: dict[str, list[JumpRecord]] = {}
for rec in records:
fname = str(rec.get("file", ""))
by_file.setdefault(fname, []).append(rec)
modified: set[str] = set()
for fname, file_records in by_file.items():
fpath = ssh_dir / fname
if not fpath.exists():
continue
file_records.sort(key=lambda r: int(r.get("line", 0)), reverse=True)
lines = fpath.read_text().splitlines(keepends=False)
if any(_PROXYJUMP_RE.match(ln) for ln in lines):
continue
for rec in file_records:
lineno = int(rec.get("line", 0))
text = str(rec.get("text", ""))
if lineno < len(lines):
lines.insert(lineno, text)
else:
lines.append(text)
fpath.write_text("\n".join(lines) + "\n")
modified.add(fname)
(ssh_dir / _SAVE_FILE).unlink(missing_ok=True)
return {"status": "ok", "count": len(records), "files": sorted(modified)}
def status(ssh_dir: Path) -> JumpStatusResult:
hosts = list_hosts(ssh_dir)
saved = load_state(ssh_dir)
return JumpStatusResult(
hosts=hosts,
count=len(hosts),
saved=saved is not None,
)
+227
View File
@@ -0,0 +1,227 @@
from __future__ import annotations
import subprocess
from pathlib import Path
SERVICE_DIR = Path("~/.config/systemd/user").expanduser()
def _run_systemctl(args: list[str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["systemctl", "--user", *args],
capture_output=True,
text=True,
timeout=30,
)
def agent_status() -> dict[str, str]:
"""Check ssh-agent systemd service status."""
result = _run_systemctl(["is-active", "ssh-agent.service"])
return {"service": "ssh-agent", "status": result.stdout.strip() or "inactive"}
def agent_start() -> dict[str, str]:
"""Enable and start ssh-agent systemd service."""
_write_agent_service()
_run_systemctl(["daemon-reload"])
_run_systemctl(["enable", "ssh-agent.service"])
_run_systemctl(["restart", "ssh-agent.service"])
return agent_status()
def agent_stop() -> dict[str, str]:
_run_systemctl(["stop", "ssh-agent.service"])
_run_systemctl(["disable", "ssh-agent.service"])
return {"service": "ssh-agent", "status": "stopped"}
def _write_agent_service() -> Path:
SERVICE_DIR.mkdir(parents=True, exist_ok=True)
path = SERVICE_DIR / "ssh-agent.service"
content = """\
[Unit]
Description=OpenSSH agent (user session)
Documentation=man:ssh-agent(1)
After=network.target
[Service]
Type=simple
ExecStart=/usr/bin/ssh-agent -D -a "${XDG_RUNTIME_DIR}/ssh-agent.socket"
ExecReload=/usr/bin/ssh-agent -k
ExecStop=/usr/bin/ssh-agent -k
Restart=on-failure
RestartSec=5
[Install]
WantedBy=default.target
"""
path.write_text(content)
return path
def timer_status() -> list[dict[str, str]]:
"""Status of all sshctl-managed timers."""
timers = [
("ssh-cleanup", "Stale SSH control socket cleanup"),
("sshctl-check", "Weekly SSH connectivity check"),
]
results = []
for name, desc in timers:
result = _run_systemctl(["is-active", f"{name}.timer"])
results.append(
{
"timer": name,
"description": desc,
"status": result.stdout.strip() or "inactive",
}
)
return results
def timer_enable(name: str) -> dict[str, str]:
valid = {"cleanup": "ssh-cleanup", "check": "sshctl-check"}
service_name = valid.get(name)
if not service_name:
return {"error": f"Unknown timer '{name}'. Valid: {', '.join(valid)}"}
_write_timer_units()
_run_systemctl(["daemon-reload"])
_run_systemctl(["enable", f"{service_name}.timer"])
_run_systemctl(["start", f"{service_name}.timer"])
result = _run_systemctl(["is-active", f"{service_name}.timer"])
return {"timer": name, "status": result.stdout.strip() or "inactive"}
def timer_disable(name: str) -> dict[str, str]:
valid = {"cleanup": "ssh-cleanup", "check": "sshctl-check"}
service_name = valid.get(name)
if not service_name:
return {"error": f"Unknown timer '{name}'"}
_run_systemctl(["stop", f"{service_name}.timer"])
_run_systemctl(["disable", f"{service_name}.timer"])
return {"timer": name, "status": "stopped"}
def _write_timer_units() -> None:
SERVICE_DIR.mkdir(parents=True, exist_ok=True)
cleanup_service = SERVICE_DIR / "ssh-cleanup.service"
cleanup_service.write_text("""\
[Unit]
Description=Stale SSH control socket cleanup
[Service]
Type=oneshot
ExecStart={sshctl} cleanup
Nice=19
IOSchedulingClass=idle
""")
cleanup_timer = SERVICE_DIR / "ssh-cleanup.timer"
cleanup_timer.write_text("""\
[Unit]
Description=Daily stale SSH control socket cleanup
[Timer]
OnCalendar=daily
Persistent=true
[Install]
WantedBy=timers.target
""")
check_service = SERVICE_DIR / "sshctl-check.service"
check_service.write_text("""\
[Unit]
Description=Weekly SSH connectivity check via sshctl
[Service]
Type=oneshot
ExecStart={sshctl} check --timeout 15
Nice=19
IOSchedulingClass=idle
""")
check_timer = SERVICE_DIR / "sshctl-check.timer"
check_timer.write_text("""\
[Unit]
Description=Weekly SSH connectivity check
[Timer]
OnCalendar=Sun 07:00
Persistent=true
[Install]
WantedBy=timers.target
""")
def init_git(ssh_dir: str | Path) -> dict[str, str]:
"""Initialize git repo in ssh_dir for config tracking."""
ssh_dir = Path(ssh_dir)
git_dir = ssh_dir / ".git"
if git_dir.exists():
return {"status": "exists", "path": str(ssh_dir)}
result = subprocess.run(
["git", "init"],
cwd=ssh_dir,
capture_output=True,
text=True,
timeout=15,
)
if result.returncode != 0:
return {"status": "error", "message": result.stderr.strip()}
_write_gitignore(ssh_dir)
return {"status": "created", "path": str(ssh_dir)}
def _write_gitignore(ssh_dir: Path) -> None:
gitignore = ssh_dir / ".gitignore"
if gitignore.exists():
return
gitignore.write_text("""\
keys.d/
keys.a/
keys.pub/
control.d/
known_hosts
known_hosts.old
report/
inventory.d/
agent/
ssh_backup_*/
scripts/*.log
.DS_Store
Thumbs.db
""")
def shell_hook(shell: str) -> dict[str, str]:
"""Print shell hook config for SSH_AUTH_SOCK."""
if shell == "zsh":
return {
"shell": shell,
"line": 'export SSH_AUTH_SOCK="${XDG_RUNTIME_DIR}/ssh-agent.socket"',
}
if shell == "bash":
return {
"shell": shell,
"line": 'export SSH_AUTH_SOCK="${XDG_RUNTIME_DIR}/ssh-agent.socket"',
}
return {"shell": shell, "line": ""}
+86
View File
@@ -0,0 +1,86 @@
from __future__ import annotations
from pathlib import Path
import pytest
from sshctl.core.config import SshctlConfig
from sshctl.core.models import HostBlock
@pytest.fixture
def ssh_dir(tmp_path: Path) -> Path:
d = tmp_path / ".ssh"
d.mkdir()
return d
@pytest.fixture
def sample_config(ssh_dir: Path) -> Path:
content = """Host *
User root
Port 22
StrictHostKeyChecking accept-new
Host server1
HostName 10.0.0.1
User admin
Port 2222
Host server2
HostName 10.0.0.2
ProxyJump server1
Host *.example.com
HostName example.com
User deploy
"""
path = ssh_dir / "config"
path.write_text(content)
return path
@pytest.fixture
def config_with_includes(ssh_dir: Path) -> Path:
conf_d = ssh_dir / "conf.d"
conf_d.mkdir()
jumps = conf_d / "98-jumpservers.conf"
jumps.write_text("""Host gw
HostName 10.0.0.254
User jumpuser
""")
creds = conf_d / "99-cred.conf"
creds.write_text("""Host *
IdentityFile ~/.ssh/keys.d/fallback
""")
main = ssh_dir / "config"
main.write_text("""Host *
User root
Port 22
Include conf.d/98-jumpservers.conf
Include conf.d/99-cred.conf
Host app1
HostName 10.0.0.10
ProxyJump gw
""")
return main
@pytest.fixture
def sample_host() -> HostBlock:
host = HostBlock(alias="test-box")
host.set("HostName", "10.0.0.1")
host.set("User", "admin")
host.set("Port", "2222")
host.set("ProxyJump", "gw")
return host
@pytest.fixture
def cfg() -> SshctlConfig:
return SshctlConfig()
+119
View File
@@ -0,0 +1,119 @@
from __future__ import annotations
from sshctl.config.resolver import ResolvedHost
from sshctl.operations.checker import HostCheck, check_host
def test_host_check_dataclass() -> None:
h = HostCheck(
alias="test",
hostname="10.0.0.1",
user="admin",
port=22,
proxy_jump="",
)
assert h.alias == "test"
assert h.hostname == "10.0.0.1"
assert h.user == "admin"
assert h.port == 22
def test_host_check_passed_both_ok() -> None:
h = HostCheck(
alias="test",
hostname="10.0.0.1",
user="admin",
port=22,
proxy_jump="",
)
assert not h.passed
h.dns_ok = True
assert not h.passed
h.tcp_ok = True
assert h.passed
def test_host_check_summary_all_ok() -> None:
h = HostCheck(
alias="test",
hostname="10.0.0.1",
user="admin",
port=22,
proxy_jump="",
)
h.dns_ok = True
h.tcp_ok = True
h.ssh_ok = True
assert h.summary == "DNS/TCP/SSH"
def test_host_check_summary_dns_fail() -> None:
h = HostCheck(
alias="test",
hostname="10.0.0.1",
user="admin",
port=22,
proxy_jump="",
)
h.dns_error = "timeout"
assert h.summary == "DNS!/TCP!/---"
def test_host_check_summary_ssh_untested() -> None:
h = HostCheck(
alias="test",
hostname="10.0.0.1",
user="admin",
port=22,
proxy_jump="",
)
h.dns_ok = True
h.tcp_ok = True
assert h.summary == "DNS/TCP/---"
def test_host_check_summary_ssh_error() -> None:
h = HostCheck(
alias="test",
hostname="10.0.0.1",
user="admin",
port=22,
proxy_jump="",
)
h.dns_ok = True
h.tcp_ok = True
h.ssh_error = "refused"
assert h.summary == "DNS/TCP/SSH!"
async def test_check_host_dns_failure() -> None:
host = ResolvedHost(
alias="nonexistent",
hostname="192.0.2.999",
user="root",
port=22,
proxy_jump="",
identity_file="",
options={},
)
result = await check_host(host, timeout=1)
assert isinstance(result.rtt_ms, float)
assert isinstance(result.dns_ok, bool)
assert isinstance(result.tcp_ok, bool)
async def test_check_host_constructs_correctly() -> None:
host = ResolvedHost(
alias="test-box",
hostname="192.0.2.1",
user="deploy",
port=2222,
proxy_jump="gw",
identity_file="",
options={},
)
result = await check_host(host, timeout=2)
assert result.alias == "test-box"
assert result.hostname == "192.0.2.1"
assert result.port == 2222
assert result.rtt_ms >= 0
+91
View File
@@ -0,0 +1,91 @@
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
from sshctl.operations.cleanup import _is_socket_in_use, cleanup_control_sockets
def test_cleanup_no_control_dir(ssh_dir: Path) -> None:
result = cleanup_control_sockets(ssh_dir)
assert result == {"total": 0, "removed": 0, "active": 0}
def test_cleanup_empty(ssh_dir: Path) -> None:
control = ssh_dir / "control.d"
control.mkdir()
result = cleanup_control_sockets(ssh_dir)
assert result == {"total": 0, "removed": 0, "active": 0}
def test_cleanup_removes_stale(ssh_dir: Path) -> None:
control = ssh_dir / "control.d"
control.mkdir()
(control / "abc.sock").write_text("")
(control / "def.sock").write_text("")
with patch("sshctl.operations.cleanup._is_socket_in_use", return_value=False):
result = cleanup_control_sockets(ssh_dir)
assert result["total"] == 2
assert result["removed"] == 2
assert result["active"] == 0
assert not (control / "abc.sock").exists()
assert not (control / "def.sock").exists()
def test_cleanup_dry_run_keeps_sockets(ssh_dir: Path) -> None:
control = ssh_dir / "control.d"
control.mkdir()
sock = control / "abc.sock"
sock.write_text("")
with patch("sshctl.operations.cleanup._is_socket_in_use", return_value=False):
result = cleanup_control_sockets(ssh_dir, dry_run=True)
assert result["total"] == 1
assert result["removed"] == 1
assert sock.exists()
def test_cleanup_active_not_removed(ssh_dir: Path) -> None:
control = ssh_dir / "control.d"
control.mkdir()
(control / "abc.sock").write_text("")
(control / "def.sock").write_text("")
def fake_in_use(sock: Path) -> bool:
return sock.name == "abc.sock"
with patch("sshctl.operations.cleanup._is_socket_in_use", side_effect=fake_in_use):
result = cleanup_control_sockets(ssh_dir)
assert result["total"] == 2
assert result["active"] == 1
assert result["removed"] == 1
assert (control / "abc.sock").exists()
assert not (control / "def.sock").exists()
def test_is_socket_in_use_with_ssh() -> None:
with patch(
"sshctl.operations.cleanup.subprocess.run",
) as mock_run:
mock_run.return_value.stdout = " 123 ssh user cwd DIR /tmp/socket"
assert _is_socket_in_use(Path("/tmp/x.sock")) is True
def test_is_socket_in_use_no_ssh() -> None:
with patch(
"sshctl.operations.cleanup.subprocess.run",
) as mock_run:
mock_run.return_value.stdout = " 123 vim user cwd DIR /tmp"
assert _is_socket_in_use(Path("/tmp/x.sock")) is False
def test_is_socket_in_use_error() -> None:
with patch(
"sshctl.operations.cleanup.subprocess.run",
side_effect=FileNotFoundError,
):
assert _is_socket_in_use(Path("/tmp/x.sock")) is False
+32
View File
@@ -0,0 +1,32 @@
from __future__ import annotations
from pathlib import Path
from sshctl.core.config import SshctlConfig, load_config
def test_default_config() -> None:
config = SshctlConfig()
assert config.ssh_dir == Path("~/.ssh").expanduser()
assert config.config_file == "config"
assert config.verbose is False
assert config.timeout == 10
def test_config_env_prefix() -> None:
assert SshctlConfig.model_config["env_prefix"] == "SSHCTL_"
def test_ssh_config_path() -> None:
config = SshctlConfig(ssh_dir=Path("/tmp/.ssh"))
assert config.ssh_config_path() == Path("/tmp/.ssh/config")
def test_conf_d_path() -> None:
config = SshctlConfig(ssh_dir=Path("/tmp/.ssh"))
assert config.conf_d_path() == Path("/tmp/.ssh/conf.d")
def test_load_config_nonexistent() -> None:
config = load_config(Path("/nonexistent/sshctl.yaml"))
assert isinstance(config, SshctlConfig)
+293
View File
@@ -0,0 +1,293 @@
from __future__ import annotations
from pathlib import Path
from sshctl.operations.configops import (
_clean_blank_lines,
find_host_file,
get_host_aliases,
get_host_star_block,
heal_config,
remove_host_from_file,
set_host_star_option,
unset_host_star_option,
validate_options,
)
STAR_CONFIG = """Host *
User root
Port 22
StrictHostKeyChecking accept-new
Include conf.d/
"""
def _write(path: Path, text: str) -> Path:
path.write_text(text)
return path
def test_get_host_star_block(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", STAR_CONFIG)
opts = get_host_star_block(cfg)
assert opts["User"] == "root"
assert opts["Port"] == "22"
assert opts["StrictHostKeyChecking"] == "accept-new"
def test_get_host_star_block_no_star(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host web\n HostName 10.0.0.1\n")
assert get_host_star_block(cfg) == {}
def test_get_host_star_block_comments_skipped(tmp_path: Path) -> None:
cfg = _write(
tmp_path / "config",
"Host *\n ## comment\n User root\n # another\n Port 22\n",
)
opts = get_host_star_block(cfg)
assert opts == {"User": "root", "Port": "22"}
def test_get_host_star_block_stops_at_include(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", STAR_CONFIG)
opts = get_host_star_block(cfg)
assert "after_include" not in opts
def test_get_host_star_block_stripped_quotes(tmp_path: Path) -> None:
cfg = _write(
tmp_path / "config",
"Host *\n IdentityFile \"~/.ssh/id_ed25519\"\n ProxyJump 'gw'\n",
)
opts = get_host_star_block(cfg)
assert opts["IdentityFile"] == "~/.ssh/id_ed25519"
assert opts["ProxyJump"] == "gw"
def test_set_host_star_option_update_existing(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", STAR_CONFIG)
set_host_star_option(cfg, "Port", "2222")
opts = get_host_star_block(cfg)
assert opts["Port"] == "2222"
text = cfg.read_text()
assert "Port 2222" in text
def test_set_host_star_option_add_new(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n User root\n\nInclude conf.d/\n")
set_host_star_option(cfg, "Port", "2222")
opts = get_host_star_block(cfg)
assert opts["Port"] == "2222"
def test_set_host_star_option_case_insensitive(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n port 22\n")
set_host_star_option(cfg, "Port", "2222")
opts = get_host_star_block(cfg)
assert opts["Port"] == "2222"
def test_set_host_star_option_no_star_block(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host web\n HostName 10.0.0.1\n")
set_host_star_option(cfg, "Port", "22")
assert "Port 22" in cfg.read_text()
def test_unset_host_star_option(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", STAR_CONFIG)
assert unset_host_star_option(cfg, "Port") is True
assert "Port" not in get_host_star_block(cfg)
assert "Port 22" not in cfg.read_text()
def test_unset_host_star_option_missing(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n User root\n")
assert unset_host_star_option(cfg, "Port") is False
def test_unset_host_star_option_case_insensitive(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n port 22\n")
assert unset_host_star_option(cfg, "Port") is True
def test_validate_options_valid(tmp_path: Path) -> None:
cfg = _write(
tmp_path / "config",
"Host *\n User root\n Port 22\n StrictHostKeyChecking accept-new\n",
)
assert validate_options(cfg) == []
def test_validate_options_unknown(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n FooBarBaz yes\n")
issues = validate_options(cfg)
assert len(issues) == 1
assert issues[0]["type"] == "unknown"
assert issues[0]["key"] == "FooBarBaz"
def test_validate_options_typo(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n hostname example.com\n")
issues = validate_options(cfg)
assert len(issues) == 1
assert issues[0]["type"] == "typo"
assert issues[0]["fixed"] == "HostName"
def test_validate_options_invalid(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n GSSAPIKeyExchange yes\n")
issues = validate_options(cfg)
assert len(issues) == 1
assert issues[0]["type"] == "invalid"
def test_validate_options_skips_comments_hosts_includes(tmp_path: Path) -> None:
cfg = _write(
tmp_path / "config",
"# comment\nHost *\n User root\nInclude conf.d/\n",
)
assert validate_options(cfg) == []
def test_validate_options_multiword_typo(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n strict hostkeychecking yes\n")
issues = validate_options(cfg)
assert len(issues) == 1
assert issues[0]["type"] == "unknown"
def test_validate_proper_casing_not_flagged(tmp_path: Path) -> None:
cfg = _write(
tmp_path / "config",
"Host *\n HostName example.com\n User root\n Port 22\n",
)
assert validate_options(cfg) == []
def test_heal_config_dry_run(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n hostname example.com\n")
actions = heal_config(cfg, auto_fix=False)
assert len(actions) == 1
assert actions[0]["action"] == "detected"
def test_heal_config_no_issues(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n User root\n")
assert heal_config(cfg, auto_fix=True) == []
def test_heal_config_fix_typo(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host web\n hostname example.com\n")
actions = heal_config(cfg, auto_fix=True)
assert any(a["action"] == "fixed" for a in actions)
assert "HostName example.com" in cfg.read_text()
def test_heal_config_keep_correct_case(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host web\n HostName example.com\n")
actions = heal_config(cfg, auto_fix=True)
assert all(a["action"] != "fixed" for a in actions)
def test_heal_config_remove_invalid(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n GSSAPIKeyExchange yes\n")
actions = heal_config(cfg, auto_fix=True)
assert any(a["action"] == "removed" for a in actions)
assert "GSSAPIKeyExchange" not in cfg.read_text()
def test_heal_config_warnings_remaining(tmp_path: Path) -> None:
cfg = _write(tmp_path / "config", "Host *\n FooBarBaz yes\n")
actions = heal_config(cfg, auto_fix=True)
assert any(a["action"] == "warning" for a in actions)
def test_find_host_file(ssh_dir: Path) -> None:
conf_d = ssh_dir / "conf.d"
conf_d.mkdir()
(conf_d / "10-servers.conf").write_text(
"Host web01\n HostName 10.0.0.1\n\nHost web02\n HostName 10.0.0.2\n"
)
found = find_host_file("web02", ssh_dir)
assert found is not None
assert found.name == "10-servers.conf"
def test_find_host_file_missing(ssh_dir: Path) -> None:
conf_d = ssh_dir / "conf.d"
conf_d.mkdir()
(conf_d / "10-servers.conf").write_text("Host web01\n HostName 10.0.0.1\n")
assert find_host_file("nope", ssh_dir) is None
def test_find_host_file_no_conf_d(ssh_dir: Path) -> None:
assert find_host_file("web01", ssh_dir) is None
def test_find_host_file_multi_alias_line(ssh_dir: Path) -> None:
conf_d = ssh_dir / "conf.d"
conf_d.mkdir()
(conf_d / "10-servers.conf").write_text("Host web01 web02\n HostName 10.0.0.1\n")
assert find_host_file("web02", ssh_dir) is not None
def test_get_host_aliases(ssh_dir: Path) -> None:
conf_d = ssh_dir / "conf.d"
conf_d.mkdir()
(conf_d / "10-servers.conf").write_text(
"Host *\n User root\nHost web01\n HostName 10.0.0.1\nHost web02\n HostName 10.0.0.2\n"
)
assert get_host_aliases(ssh_dir) == ["web01", "web02"]
def test_get_host_aliases_no_conf_d(ssh_dir: Path) -> None:
assert get_host_aliases(ssh_dir) == []
def test_get_host_aliases_recursive(ssh_dir: Path) -> None:
nested = ssh_dir / "conf.d" / "internal" / "prod"
nested.mkdir(parents=True)
(nested / "01-web.conf").write_text("Host app01\n HostName 10.0.0.1\n")
assert get_host_aliases(ssh_dir) == ["app01"]
def test_remove_host_from_file(tmp_path: Path) -> None:
f = _write(
tmp_path / "hosts.conf",
"Host web01\n HostName 10.0.0.1\n\nHost web02\n HostName 10.0.0.2\n",
)
assert remove_host_from_file("web01", f) is True
text = f.read_text()
assert "web01" not in text
assert "web02" in text
def test_remove_host_from_file_missing(tmp_path: Path) -> None:
f = _write(tmp_path / "hosts.conf", "Host web01\n HostName 10.0.0.1\n")
assert remove_host_from_file("nope", f) is False
def test_remove_host_cleans_duplicate_blanks(tmp_path: Path) -> None:
f = _write(
tmp_path / "hosts.conf",
"Host web01\n HostName 10.0.0.1\n\n\n\n\nHost web02\n HostName 10.0.0.2\n",
)
remove_host_from_file("web01", f)
text = f.read_text()
assert "\n\n\n" not in text
def test_remove_host_keeps_comments_before_next(tmp_path: Path) -> None:
f = _write(
tmp_path / "hosts.conf",
"Host web01\n HostName 10.0.0.1\n\n# next host\nHost web02\n HostName 10.0.0.2\n",
)
remove_host_from_file("web01", f)
assert "web02" in f.read_text()
def test_clean_blank_lines() -> None:
cleaned = _clean_blank_lines(["Host a", "", "", " HostName x", ""])
assert cleaned == ["Host a", "", " HostName x", ""]
+64
View File
@@ -0,0 +1,64 @@
from __future__ import annotations
from pathlib import Path
from sshctl.operations.events import Event, flush, log_event, recent
def test_event_to_dict() -> None:
ev = Event(
command="host list",
status="ok",
args={"--verbose": "true"},
result="",
duration_ms=12.5,
)
d = ev.to_dict()
assert d["cmd"] == "host list"
assert d["status"] == "ok"
assert d["args"] == {"--verbose": "true"}
assert d["duration_ms"] == 12.5
assert isinstance(d["ts"], float)
def test_log_event_and_flush(ssh_dir: Path) -> None:
log_event(Event(command="check", status="ok", duration_ms=1.5))
flush(ssh_dir)
rows = recent(ssh_dir)
assert len(rows) == 1
assert rows[0]["cmd"] == "check"
assert rows[0]["status"] == "ok"
def test_flush_empty_no_file(ssh_dir: Path) -> None:
flush(ssh_dir)
assert recent(ssh_dir) == []
def test_recent_empty_when_no_file(ssh_dir: Path) -> None:
assert recent(ssh_dir) == []
def test_recent_limits(ssh_dir: Path) -> None:
for i in range(10):
log_event(Event(command=f"cmd{i}", status="ok"))
flush(ssh_dir)
rows = recent(ssh_dir, limit=3)
assert len(rows) == 3
assert rows[0]["cmd"] == "cmd7"
def test_recent_skips_bad_json(ssh_dir: Path) -> None:
report = ssh_dir / "report"
report.mkdir(exist_ok=True)
(report / "events.ndjson").write_text("not-json\n")
assert recent(ssh_dir) == []
def test_flush_writes_valid_ndjson(ssh_dir: Path) -> None:
log_event(Event(command="tree", status="ok"))
log_event(Event(command="diff", status="error"))
flush(ssh_dir)
path = ssh_dir / "report" / "events.ndjson"
lines = path.read_text().strip().splitlines()
assert len(lines) == 2
+25
View File
@@ -0,0 +1,25 @@
from __future__ import annotations
from sshctl.core.exceptions import (
ConfigError,
ParseError,
ResolverError,
SshctlError,
)
def test_exception_hierarchy() -> None:
assert issubclass(ConfigError, SshctlError)
assert issubclass(ParseError, SshctlError)
assert issubclass(ResolverError, SshctlError)
def test_sshctl_error_is_base() -> None:
assert issubclass(SshctlError, Exception)
def test_exception_raised() -> None:
try:
raise ConfigError("test error")
except SshctlError as e:
assert str(e) == "test error"
+24
View File
@@ -0,0 +1,24 @@
from __future__ import annotations
from sshctl.operations.executor import ExecResult
def test_exec_result_dataclass() -> None:
r = ExecResult(alias="server1", returncode=0, stdout="ok", stderr="")
assert r.alias == "server1"
assert r.returncode == 0
assert r.stdout == "ok"
assert r.error == ""
def test_exec_result_error() -> None:
r = ExecResult(alias="server1", returncode=None, error="Timeout")
assert r.returncode is None
assert r.error == "Timeout"
def test_exec_result_defaults() -> None:
r = ExecResult(alias="test", returncode=0)
assert r.stdout == ""
assert r.stderr == ""
assert r.error == ""
+67
View File
@@ -0,0 +1,67 @@
from __future__ import annotations
from pathlib import Path
from sshctl.config.resolver import ResolvedHost
from sshctl.export.exporter import export_csv, export_csv_to_file
def test_export_csv_with_hosts() -> None:
hosts = [
ResolvedHost(
alias="server1",
hostname="10.0.0.1",
user="admin",
port=22,
proxy_jump="gw",
identity_file="~/.ssh/id_rsa",
options={},
),
ResolvedHost(
alias="server2",
hostname="10.0.0.2",
user="root",
port=2222,
proxy_jump="",
identity_file="",
options={},
),
]
csv_content = export_csv(hosts)
lines = csv_content.strip().splitlines()
assert len(lines) == 3
assert lines[0] == "Host,HostName,User,Port,ProxyJump,IdentityFile"
assert "server1" in lines[1]
assert "server2" in lines[2]
assert "10.0.0.1" in lines[1]
assert "gw" in lines[1]
def test_export_csv_to_file(tmp_path: Path) -> None:
hosts = [
ResolvedHost(
alias="test",
hostname="10.0.0.1",
user="root",
port=22,
proxy_jump="",
identity_file="",
options={},
),
]
path = export_csv_to_file(tmp_path / "output.csv", hosts)
assert path.exists()
content = path.read_text()
assert "test" in content
assert "10.0.0.1" in content
def test_export_csv_requires_args() -> None:
raised = False
try:
export_csv()
except ValueError:
raised = True
assert raised, "Expected ValueError"
+103
View File
@@ -0,0 +1,103 @@
from __future__ import annotations
from pathlib import Path
from sshctl.config.generator import generate_config_text, generate_host_block, write_host_file
from sshctl.config.parser import parse_config
from sshctl.core.models import HostBlock, SshConfig
def test_generate_host_block() -> None:
host = HostBlock(alias="test")
host.set("HostName", "10.0.0.1")
host.set("User", "admin")
text = generate_host_block(host)
assert "Host test" in text
assert "HostName 10.0.0.1" in text
assert "User admin" in text
def test_generate_config_roundtrip() -> None:
original = """Host *
User root
Port 22
Host server1
HostName 10.0.0.1
User admin
Port 2222
Include conf.d/98-jumpservers.conf
"""
parsed = parse_config(original)
generated = generate_config_text(parsed, include_header=False)
reparsed = parse_config(generated)
assert len(reparsed.blocks) == len(parsed.blocks)
assert reparsed.blocks[0].alias == parsed.blocks[0].alias
assert reparsed.blocks[1].hostname == parsed.blocks[1].hostname
assert reparsed.blocks[1].user == parsed.blocks[1].user
def test_generate_config_header() -> None:
parsed = parse_config("Host test\n HostName 10.0.0.1\n")
text = generate_config_text(parsed, include_header=True)
assert text.startswith("# Generated by sshctl")
def test_generate_empty() -> None:
config = SshConfig()
text = generate_config_text(config)
assert "# Generated by sshctl" in text
def test_write_host_file(tmp_path: Path) -> None:
hosts = [
HostBlock(alias="web01", options={"HostName": "10.0.0.1"}),
HostBlock(alias="web02", options={"HostName": "10.0.0.2"}),
]
path = write_host_file(hosts, tmp_path / "test.conf")
assert path.exists()
content = path.read_text()
assert "Host web01" in content
assert "Host web02" in content
def test_option_ordering() -> None:
host = HostBlock(alias="test")
host.set("HostName", "10.0.0.1")
host.set("User", "admin")
host.set("Port", "2222")
host.set("ProxyJump", "gw")
text = generate_host_block(host)
lines = text.splitlines()
hostname_idx = next(i for i, line in enumerate(lines) if "HostName" in line)
port_idx = next(i for i, line in enumerate(lines) if "Port" in line)
proxy_idx = next(i for i, line in enumerate(lines) if "ProxyJump" in line)
assert hostname_idx < proxy_idx
assert port_idx < proxy_idx
def test_roundtrip_preserves_options() -> None:
config = SshConfig()
config.raw_includes.append("conf.d/*.conf")
block = HostBlock(alias="app")
block.set("HostName", "app.example.com")
block.set("User", "deploy")
block.set("Port", "2222")
block.set("ProxyJump", "gw")
block.set("ForwardAgent", "yes")
config.blocks.append(block)
text = generate_config_text(config, include_header=False)
reparsed = parse_config(text)
reblock = reparsed.find("app")
assert reblock is not None
assert reblock.hostname == "app.example.com"
assert reblock.user == "deploy"
assert reblock.proxy_jump == "gw"
assert reblock.get("ForwardAgent") == "yes"
+31
View File
@@ -0,0 +1,31 @@
from __future__ import annotations
from pathlib import Path
from sshctl.export.inventory import _to_group_name, generate_inventory
def test_to_group_name_simple() -> None:
assert _to_group_name("internal") == "internal"
assert _to_group_name("cluster1") == "cluster1"
def test_to_group_name_with_hyphens() -> None:
assert _to_group_name("8xx-serverfarm") == "z8xx_serverfarm"
def test_to_group_name_starts_with_digit() -> None:
assert _to_group_name("8xx") == "z8xx"
def test_generate_inventory_no_ssh_dir(tmp_path: Path) -> None:
result = generate_inventory(tmp_path / "nonexistent")
assert result == {}
def test_generate_inventory_empty_config(tmp_path: Path) -> None:
ssh_dir = tmp_path / ".ssh"
ssh_dir.mkdir()
(ssh_dir / "config").write_text("Host *\n User root\n")
result = generate_inventory(ssh_dir)
assert result == {}
+271
View File
@@ -0,0 +1,271 @@
from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import patch
from sshctl.operations.jump import (
add_network,
current_ssid,
is_internal_by_ssid,
is_internal_by_subnet,
known_networks,
list_hosts,
load_state,
off,
on,
save_state,
status,
)
def _mk_conf(ssh_dir: Path, name: str, text: str) -> Path:
conf_d = ssh_dir / "conf.d"
conf_d.mkdir(parents=True, exist_ok=True)
p = conf_d / name
p.write_text(text)
return p
def test_known_networks_defaults(ssh_dir: Path) -> None:
assert known_networks(ssh_dir) == ["MobinNet*", "Mobin*"]
def test_known_networks_custom(ssh_dir: Path) -> None:
(ssh_dir / ".internal_ssids").write_text("Home*\n # comment\nOffice\n")
assert known_networks(ssh_dir) == ["Home*", "Office"]
def test_add_network_creates_file(ssh_dir: Path) -> None:
add_network(ssh_dir, "MyWiFi")
lines = (ssh_dir / ".internal_ssids").read_text().splitlines()
assert lines[-1] == "MyWiFi"
def test_add_network_does_not_duplicate(ssh_dir: Path) -> None:
add_network(ssh_dir, "TestNet")
add_network(ssh_dir, "TestNet")
assert (ssh_dir / ".internal_ssids").read_text().strip().count("TestNet") == 1
def test_current_ssid_detected() -> None:
with patch("sshctl.operations.jump.subprocess.run") as m:
m.return_value.stdout = "yes:MyWiFi\nno:OldNet\n"
assert current_ssid() == "MyWiFi"
def test_current_ssid_none() -> None:
with patch("sshctl.operations.jump.subprocess.run") as m:
m.return_value.stdout = "no:OldNet\n"
assert current_ssid() is None
def test_current_ssid_subprocess_error() -> None:
with patch("sshctl.operations.jump.subprocess.run", side_effect=FileNotFoundError):
assert current_ssid() is None
def test_is_internal_by_ssid_no_wifi(tmp_path: Path) -> None:
with patch("sshctl.operations.jump.current_ssid", return_value=None):
assert is_internal_by_ssid(tmp_path) is None
def test_is_internal_by_ssid_match(tmp_path: Path) -> None:
(tmp_path / ".internal_ssids").write_text("MobinNet*\n")
with patch("sshctl.operations.jump.current_ssid", return_value="MobinNet-5G"):
assert is_internal_by_ssid(tmp_path) is True
def test_is_internal_by_ssid_no_match(tmp_path: Path) -> None:
(tmp_path / ".internal_ssids").write_text("MobinNet*\n")
with patch("sshctl.operations.jump.current_ssid", return_value="CoffeeShop"):
assert is_internal_by_ssid(tmp_path) is False
def test_is_internal_by_ssid_default_patterns(tmp_path: Path) -> None:
with patch("sshctl.operations.jump.current_ssid", return_value="Mobin-Office"):
assert is_internal_by_ssid(tmp_path) is True
def test_list_hosts(ssh_dir: Path) -> None:
_mk_conf(
ssh_dir,
"10-servers.conf",
"Host gw\n HostName 10.0.0.254\n\nHost web01\n HostName 10.0.0.1\n ProxyJump gw\n",
)
hosts = list_hosts(ssh_dir)
assert len(hosts) == 1
assert hosts[0]["alias"] == "web01"
assert hosts[0]["jump"] == "gw"
assert hosts[0]["file"] == "conf.d/10-servers.conf"
def test_list_hosts_empty(ssh_dir: Path) -> None:
assert list_hosts(ssh_dir) == []
def test_list_hosts_trailing_host_with_proxy(ssh_dir: Path) -> None:
_mk_conf(
ssh_dir,
"10-servers.conf",
"Host web01\n HostName 10.0.0.1\n ProxyJump gw\n",
)
hosts = list_hosts(ssh_dir)
assert len(hosts) == 1
assert hosts[0]["alias"] == "web01"
def test_save_state(ssh_dir: Path) -> None:
_mk_conf(
ssh_dir,
"10-servers.conf",
"Host web01\n HostName 10.0.0.1\n ProxyJump gw\n",
)
records = save_state(ssh_dir)
assert len(records) == 1
assert records[0]["file"] == "conf.d/10-servers.conf"
assert records[0]["text"] == " ProxyJump gw"
assert (ssh_dir / ".proxyjump_save").exists()
def test_save_state_no_proxy(ssh_dir: Path) -> None:
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n HostName 10.0.0.1\n")
assert save_state(ssh_dir) == []
def test_load_state_missing(ssh_dir: Path) -> None:
assert load_state(ssh_dir) is None
def test_load_state_invalid_json(ssh_dir: Path) -> None:
(ssh_dir / ".proxyjump_save").write_text("{not json")
assert load_state(ssh_dir) is None
def test_load_state_valid(ssh_dir: Path) -> None:
(ssh_dir / ".proxyjump_save").write_text(
json.dumps([{"file": "conf.d/x.conf", "line": 3, "text": " ProxyJump gw"}])
)
records = load_state(ssh_dir)
assert records is not None
assert records[0]["file"] == "conf.d/x.conf"
def test_off_noop(ssh_dir: Path) -> None:
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n HostName 10.0.0.1\n")
result = off(ssh_dir)
assert result["status"] == "noop"
def test_off_removes_proxy_jump(ssh_dir: Path) -> None:
f = _mk_conf(
ssh_dir,
"10-servers.conf",
"Host web01\n HostName 10.0.0.1\n ProxyJump gw\n",
)
result = off(ssh_dir)
assert result["status"] == "ok"
assert result["count"] == 1
assert "ProxyJump" not in f.read_text()
assert (ssh_dir / ".proxyjump_save").exists()
def test_off_multiple_files(ssh_dir: Path) -> None:
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
_mk_conf(ssh_dir, "20-servers.conf", "Host web02\n ProxyJump gw\n")
result = off(ssh_dir)
assert result["status"] == "ok"
assert result["count"] == 2
assert len(result["files"]) == 2
def test_on_noop(ssh_dir: Path) -> None:
result = on(ssh_dir)
assert result["status"] == "noop"
def test_on_restores(ssh_dir: Path) -> None:
_mk_conf(
ssh_dir,
"10-servers.conf",
"Host web01\n HostName 10.0.0.1\n ProxyJump gw\n",
)
off(ssh_dir)
f = ssh_dir / "conf.d" / "10-servers.conf"
result = on(ssh_dir)
assert result["status"] == "ok"
assert "ProxyJump gw" in f.read_text()
assert not (ssh_dir / ".proxyjump_save").exists()
def test_on_restores_at_original_line(ssh_dir: Path) -> None:
conf = "Host web01\n HostName 10.0.0.1\n ProxyJump gw\n\nHost web02\n HostName 10.0.0.2\n"
_mk_conf(ssh_dir, "10-servers.conf", conf)
off(ssh_dir)
on(ssh_dir)
text = (ssh_dir / "conf.d" / "10-servers.conf").read_text()
assert text.index("ProxyJump") < text.index("web02")
def test_on_skips_file_with_existing_proxy(ssh_dir: Path) -> None:
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
off(ssh_dir)
(ssh_dir / "conf.d" / "10-servers.conf").write_text("Host web01\n ProxyJump newgw\n")
result = on(ssh_dir)
assert result["files"] == []
assert "newgw" in (ssh_dir / "conf.d" / "10-servers.conf").read_text()
def test_on_missing_file_skipped(ssh_dir: Path) -> None:
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
off(ssh_dir)
(ssh_dir / "conf.d" / "10-servers.conf").unlink()
result = on(ssh_dir)
assert result["status"] == "ok"
def test_on_line_appended_when_out_of_range(ssh_dir: Path) -> None:
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
off(ssh_dir)
(ssh_dir / "conf.d" / "10-servers.conf").write_text("Host web01\n")
result = on(ssh_dir)
assert result["status"] == "ok"
assert "ProxyJump gw" in (ssh_dir / "conf.d" / "10-servers.conf").read_text()
def test_status(ssh_dir: Path) -> None:
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
info = status(ssh_dir)
assert info["count"] == 1
assert info["saved"] is False
def test_status_saved(ssh_dir: Path) -> None:
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
off(ssh_dir)
info = status(ssh_dir)
assert info["saved"] is True
def test_is_internal_by_subnet_no_gateways(ssh_dir: Path) -> None:
reachable, probes = is_internal_by_subnet(ssh_dir)
assert reachable is False
assert probes == []
def test_is_internal_by_subnet_reachable(ssh_dir: Path) -> None:
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
probes = [{"gateway": "gw", "ip": "10.0.0.254", "port": 22, "reachable": True, "error": ""}]
with patch("sshctl.operations.jump._subnet_reachable", return_value=probes):
reachable, result = is_internal_by_subnet(ssh_dir)
assert reachable is True
assert result == probes
def test_jump_gateways_collects_unique(ssh_dir: Path) -> None:
from sshctl.operations.jump import _jump_gateways
_mk_conf(ssh_dir, "10-servers.conf", "Host a\n ProxyJump gw\n")
_mk_conf(ssh_dir, "20-servers.conf", "Host b\n ProxyJump gw:2222\n ProxyJump other\n")
assert _jump_gateways(ssh_dir) == {"gw", "other"}
+24
View File
@@ -0,0 +1,24 @@
from __future__ import annotations
import structlog
from sshctl.core.logutil import get_logger, setup_logging
def test_setup_logging_plain() -> None:
setup_logging(verbose=False)
assert structlog.is_configured() is True
logger = get_logger("test")
assert logger is not None
def test_setup_logging_verbose() -> None:
setup_logging(verbose=True)
assert structlog.is_configured() is True
def test_get_logger_cached() -> None:
setup_logging(verbose=True)
first = get_logger("cache-test")
second = get_logger("cache-test")
assert first is second
+128
View File
@@ -0,0 +1,128 @@
from __future__ import annotations
from pathlib import Path
from sshctl.core.models import HostBlock, SshConfig, ZonePath
def test_host_block_defaults() -> None:
host = HostBlock(alias="test")
assert host.alias == "test"
assert host.options == {}
assert host.hostname is None
assert host.user is None
assert host.port is None
assert host.proxy_jump is None
def test_host_block_set_get() -> None:
host = HostBlock(alias="test")
host.set("HostName", "10.0.0.1")
host.set("User", "admin")
assert host.hostname == "10.0.0.1"
assert host.user == "admin"
def test_host_block_port() -> None:
host = HostBlock(alias="test")
host.set("Port", "2222")
assert host.port == 2222
def test_host_block_port_none() -> None:
host = HostBlock(alias="test")
assert host.port is None
def test_host_block_default_get() -> None:
host = HostBlock(alias="test")
assert host.get("Nonexistent") is None
assert host.get("Nonexistent", "default") == "default"
def test_ssh_config_find() -> None:
config = SshConfig()
config.blocks.append(HostBlock(alias="server1"))
config.blocks.append(HostBlock(alias="server2"))
found = config.find("server1")
assert found is not None
assert found.alias == "server1"
assert config.find("nonexistent") is None
def test_ssh_config_aliases() -> None:
config = SshConfig()
config.blocks.append(HostBlock(alias="*"))
config.blocks.append(HostBlock(alias="server1"))
config.blocks.append(HostBlock(alias="server2"))
assert config.aliases == ["server1", "server2"]
def test_ssh_config_aliases_exclude_glob_patterns() -> None:
config = SshConfig()
config.blocks.append(HostBlock(alias="web01"))
config.blocks.append(HostBlock(alias="192.168.*"))
config.blocks.append(HostBlock(alias="10.*"))
config.blocks.append(HostBlock(alias="203.0.113.* 198.51.100.*"))
config.blocks.append(HostBlock(alias="web[0-9]"))
config.blocks.append(HostBlock(alias="db?"))
config.blocks.append(HostBlock(alias="*"))
assert config.aliases == ["web01"]
def test_ssh_config_find_all() -> None:
config = SshConfig()
config.blocks.append(HostBlock(alias="web01"))
config.blocks.append(HostBlock(alias="web02"))
config.blocks.append(HostBlock(alias="db01"))
result = config.find_all("web*")
assert len(result) == 2
assert result[0].alias == "web01"
assert result[1].alias == "web02"
def test_zone_path_internal() -> None:
zp = ZonePath(
universe="internal",
sector="AZ",
cluster="cluster1",
zone="8xx-serverfarm",
vlan="82-k8s",
)
expected = Path("conf.d/internal/AZ/cluster1/8xx-serverfarm/82-k8s")
assert zp.to_rel_path() == expected
def test_zone_path_external() -> None:
zp = ZonePath(universe="external")
assert zp.to_rel_path() == Path("conf.d/external")
def test_zone_path_from_rel_path() -> None:
zp = ZonePath.from_rel_path(Path("conf.d/internal/AZ/cluster1"))
assert zp.universe == "internal"
assert zp.sector == "AZ"
assert zp.cluster == "cluster1"
assert zp.zone is None
def test_zone_path_invalid() -> None:
raised = False
try:
ZonePath.from_rel_path(Path("other/file"))
except ValueError:
raised = True
assert raised, "Expected ValueError"
def test_zone_path_unknown_universe() -> None:
raised = False
try:
ZonePath.from_rel_path(Path("conf.d/unknown"))
except ValueError:
raised = True
assert raised, "Expected ValueError"
+136
View File
@@ -0,0 +1,136 @@
from __future__ import annotations
from pathlib import Path
from sshctl.config.parser import parse_config, parse_file
from sshctl.core.exceptions import ParseError
def test_parse_empty_config() -> None:
config = parse_config("")
assert config.blocks == []
assert config.raw_includes == []
def test_parse_single_host() -> None:
text = """Host myserver
HostName 10.0.0.1
User admin
Port 2222
"""
config = parse_config(text)
assert len(config.blocks) == 1
block = config.blocks[0]
assert block.alias == "myserver"
assert block.hostname == "10.0.0.1"
assert block.user == "admin"
assert block.port == 2222
def test_parse_host_star() -> None:
text = """Host *
User root
Port 22
"""
config = parse_config(text)
assert len(config.blocks) == 1
assert config.blocks[0].alias == "*"
assert config.blocks[0].user == "root"
def test_parse_multiple_hosts() -> None:
text = """Host server1
HostName 10.0.0.1
Host server2
HostName 10.0.0.2
ProxyJump server1
"""
config = parse_config(text)
assert len(config.blocks) == 2
assert config.blocks[0].alias == "server1"
assert config.blocks[1].alias == "server2"
assert config.blocks[1].proxy_jump == "server1"
def test_parse_includes() -> None:
text = """Host *
User root
Include conf.d/internal/*.conf
Include conf.d/98-jumpservers.conf
"""
config = parse_config(text)
assert len(config.raw_includes) == 2
assert "conf.d/internal/*.conf" in config.raw_includes
assert "conf.d/98-jumpservers.conf" in config.raw_includes
def test_parse_comments_and_blanks() -> None:
text = """# This is a comment
Host test
HostName example.com
# indent comment
User tester
"""
config = parse_config(text)
assert len(config.blocks) == 1
assert config.blocks[0].alias == "test"
def test_parse_file_not_found() -> None:
raised = False
try:
parse_file("/nonexistent/ssh_config")
except ParseError:
raised = True
assert raised, "Expected ParseError"
def test_parse_file(sample_config: Path) -> None:
config = parse_file(sample_config)
assert len(config.blocks) >= 3
assert config.find("server1") is not None
assert config.find("server2") is not None
def test_parse_quoted_values() -> None:
text = """Host test
IdentityFile "~/.ssh/keys.d/id_ed25519"
ProxyJump 'gw.example.com'
"""
config = parse_config(text)
block = config.blocks[0]
assert block.get("IdentityFile") == "~/.ssh/keys.d/id_ed25519"
assert block.get("ProxyJump") == "gw.example.com"
def test_find_all_wildcard() -> None:
text = """Host web01
HostName 10.0.0.1
Host web02
HostName 10.0.0.2
Host db01
HostName 10.0.0.3
"""
config = parse_config(text)
webs = config.find_all("web*")
assert len(webs) == 2
def test_aliases_property() -> None:
text = """Host *
User root
Host server1
HostName 10.0.0.1
Host server2
HostName 10.0.0.2
"""
config = parse_config(text)
assert config.aliases == ["server1", "server2"]
+35
View File
@@ -0,0 +1,35 @@
from __future__ import annotations
from pathlib import Path
from sshctl.export.permissions import check_permissions, fix_ssh_permissions
def test_check_permissions_ok(tmp_path: Path) -> None:
ssh_dir = tmp_path / ".ssh"
ssh_dir.mkdir(mode=0o700)
(ssh_dir / "config").write_text("Host test\n", encoding="utf-8")
(ssh_dir / "config").chmod(0o600)
(ssh_dir / "keys.p").mkdir(mode=0o700)
(ssh_dir / "keys.p" / "key.pub").write_text("ssh-ed25519 AAA...")
(ssh_dir / "keys.p" / "key.pub").chmod(0o644)
issues = check_permissions(ssh_dir)
total = sum(len(v) for v in issues.values())
assert total == 0
def test_fix_permissions(tmp_path: Path) -> None:
ssh_dir = tmp_path / ".ssh"
ssh_dir.mkdir(mode=0o755)
(ssh_dir / "config").write_text("Host test\n", encoding="utf-8")
(ssh_dir / "config").chmod(0o644)
(ssh_dir / "id_rsa").write_text("private key data")
(ssh_dir / "id_rsa").chmod(0o644)
counts = fix_ssh_permissions(ssh_dir)
assert counts["dirs"] >= 1
assert counts["private"] >= 1
assert (ssh_dir / "config").stat().st_mode & 0o777 == 0o600
assert (ssh_dir / "id_rsa").stat().st_mode & 0o777 == 0o600
+101
View File
@@ -0,0 +1,101 @@
from __future__ import annotations
from pathlib import Path
import pytest
from sshctl.config.resolver import (
ResolvedHost,
_parse_ssh_g_output,
resolve_all_hosts,
resolve_host,
)
from sshctl.core.exceptions import ResolverError
def test_resolved_host_namedtuple() -> None:
host = ResolvedHost(
alias="server1",
hostname="10.0.0.1",
user="admin",
port=22,
proxy_jump="gw",
identity_file="~/.ssh/id_rsa",
options={"hostname": "10.0.0.1", "user": "admin"},
)
assert host.alias == "server1"
assert host.hostname == "10.0.0.1"
assert host.port == 22
assert host.proxy_jump == "gw"
def test_parse_ssh_g_output_basic() -> None:
text = "hostname 10.0.0.1\nuser admin\nport 2222\n"
opts = _parse_ssh_g_output(text)
assert opts["hostname"] == "10.0.0.1"
assert opts["user"] == "admin"
assert opts["port"] == "2222"
def test_parse_ssh_g_output_empty() -> None:
assert _parse_ssh_g_output("") == {}
def test_parse_ssh_g_output_blank_lines() -> None:
text = "\n\nhostname test\n\n"
opts = _parse_ssh_g_output(text)
assert opts == {"hostname": "test"}
def test_parse_ssh_g_output_first_wins() -> None:
text = "hostname first\nhostname second\n"
opts = _parse_ssh_g_output(text)
assert opts["hostname"] == "first"
def test_parse_ssh_g_output_quoted_values() -> None:
text = 'identityfile "~/.ssh/key"\n'
opts = _parse_ssh_g_output(text)
assert opts["identityfile"] == "~/.ssh/key"
def test_parse_ssh_g_output_single_word_line() -> None:
text = "invalid\nhostname test\n"
opts = _parse_ssh_g_output(text)
assert opts == {"hostname": "test"}
def test_resolve_host_ssh_not_found() -> None:
with pytest.raises(ResolverError, match="Failed to resolve"):
resolve_host("test", "/nonexistent/config", ssh_bin="ssh-nonexistent")
def test_resolve_all_hosts_empty_config(tmp_path: Path) -> None:
config_path = tmp_path / "config"
config_path.write_text("Host *\n User root\n")
results = resolve_all_hosts(config_path, ssh_bin="ssh-nonexistent")
assert results == []
def test_resolve_host_uses_correct_cmd() -> None:
import subprocess
def _fake_run(*_args: object, **_kwargs: object) -> object:
class FakeResult:
returncode = 0
stdout = "hostname 10.0.0.1\nuser testuser\nport 2222\nproxyjump jumpbox\n"
stderr = ""
return FakeResult()
original_run = subprocess.run
subprocess.run = _fake_run # type: ignore[assignment]
try:
host = resolve_host("myserver", "/tmp/fake_config")
assert host.alias == "myserver"
assert host.hostname == "10.0.0.1"
assert host.user == "testuser"
assert host.port == 2222
assert host.proxy_jump == "jumpbox"
finally:
subprocess.run = original_run
+21
View File
@@ -0,0 +1,21 @@
from __future__ import annotations
from pathlib import Path
from sshctl.config.validator import validate_config
def test_validate_nonexistent(tmp_path: Path) -> None:
result = validate_config(tmp_path / "nonexistent")
assert not result.passed
assert any("not found" in i.message for i in result.issues)
def test_validate_valid(sample_config: Path) -> None:
result = validate_config(sample_config)
assert result.passed
def test_validate_with_includes(config_with_includes: Path) -> None:
result = validate_config(config_with_includes)
assert result.passed