Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 36 additions & 33 deletions ssh_keyup.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,14 @@
from datetime import date
from pathlib import Path
from shutil import which
from typing import Dict, List, Optional, Tuple, Union
from typing import Dict, List, Optional, Set, Tuple, Union

if sys.platform == "win32":
import ctypes
import msvcrt
else:
import termios
import tty


class CLI:
Expand Down Expand Up @@ -53,29 +60,29 @@ class CLI:
S_SUCCESS = GREEN
S_STATUS = CYAN

STYLE_ATTRS = (
"BOLD", "DIM", "RESET", "GREEN", "RED", "YELLOW", "CYAN",
"HIDE_CUR", "SHOW_CUR", "S_BANNER", "S_VERSION", "S_SEPARATOR",
"S_HINT", "S_SSH_WARNING", "S_SSH_INFO", "S_SUCCESS", "S_STATUS",
)

def __init__(self) -> None:
if not sys.stdout.isatty():
CLI.BOLD = CLI.DIM = CLI.RESET = ""
CLI.GREEN = CLI.RED = CLI.YELLOW = CLI.CYAN = ""
CLI.HIDE_CUR = CLI.SHOW_CUR = ""
CLI.S_BANNER = CLI.S_VERSION = CLI.S_SEPARATOR = ""
CLI.S_HINT = CLI.S_SSH_WARNING = ""
CLI.S_SSH_INFO = CLI.S_SUCCESS = CLI.S_STATUS = ""
for attr in CLI.STYLE_ATTRS:
setattr(CLI, attr, "")

@staticmethod
def enable_ansi() -> None:
"""Enable ANSI escape sequences on Windows 10+."""
if sys.platform != "win32":
return
try:
import ctypes
k = ctypes.windll.kernel32 # type: ignore[attr-defined]
h = k.GetStdHandle(-11)
m = ctypes.c_ulong()
k.GetConsoleMode(h, ctypes.byref(m))
k.SetConsoleMode(h, m.value | 0x0004)
except Exception:
pass
if sys.platform == "win32":
try:
k = ctypes.windll.kernel32
h = k.GetStdHandle(-11)
m = ctypes.c_ulong()
k.GetConsoleMode(h, ctypes.byref(m))
k.SetConsoleMode(h, m.value | 0x0004)
except Exception:
pass

@staticmethod
def banner() -> None:
Expand Down Expand Up @@ -108,7 +115,7 @@ def fail(msg: str) -> None:
@staticmethod
def fatal(msg: str) -> None:
"""Print an error message and exit."""
cli.fail(msg)
CLI.fail(msg)
sys.exit(1)

@staticmethod
Expand Down Expand Up @@ -167,7 +174,6 @@ def msg(msg: str = "") -> None:
def _read_key() -> str:
"""Read a single keypress."""
if sys.platform == "win32":
import msvcrt
ch = msvcrt.getwch()
if ch == "\x03":
raise KeyboardInterrupt
Expand All @@ -177,8 +183,6 @@ def _read_key() -> str:
return {"K": "left", "M": "right"}.get(msvcrt.getwch(), "")
return "esc" if ch == "\x1b" else ch
else:
import termios
import tty
fd = sys.stdin.fileno()
old = termios.tcgetattr(fd)
try:
Expand Down Expand Up @@ -262,10 +266,10 @@ def _find_git_bash() -> Optional[str]:
def __init__(self) -> None:
self.git_bash = Runner._find_git_bash()
openssh = all(which(c) for c in ("ssh", "ssh-keygen"))
self.mode = (
self.mode: Optional[str] = (
"native" if openssh
else ("gitbash" if self.git_bash else None)
) # type: Optional[str]
)

def check(self) -> None:
"""Exit with guidance if no SSH tools are available."""
Expand All @@ -286,7 +290,8 @@ def _subprocess_args(
"""Prepare the command and shell flag for subprocess.run."""
if self.mode == "native":
return cmd, isinstance(cmd, str)
assert self.git_bash
if self.git_bash is None:
raise RuntimeError("gitbash mode without Git Bash path")
sh = (cmd if isinstance(cmd, str)
else " ".join(shlex.quote(a) for a in cmd))
return [self.git_bash, "-c", sh], False
Expand All @@ -311,7 +316,7 @@ class SSHConfig:
@staticmethod
def _find_managed_blocks(text: str) -> Dict[str, Tuple[int, int]]:
"""Find ssh-keyup managed blocks in SSH config text."""
blocks = {} # type: Dict[str, Tuple[int, int]]
blocks: Dict[str, Tuple[int, int]] = {}
for m in re.finditer(
r"^#ssh-keyup:begin (\S+)[^\n]*\n.*?^#ssh-keyup:end \1[^\n]*\n?",
text, re.MULTILINE | re.DOTALL,
Expand Down Expand Up @@ -516,7 +521,7 @@ def deploy(runner: Runner, user: str, host: str, pub_path: Path) -> bool:
"\nSSH connection failed. Check host and credentials."
)
if stderr.strip():
seen = set() # type: set
seen: Set[str] = set()
for line in stderr.strip().splitlines():
if line not in seen and not line.startswith("debug1:"):
seen.add(line)
Expand Down Expand Up @@ -609,17 +614,15 @@ def gather_input(args: argparse.Namespace) -> Tuple[str, str, str]:
return host, user, alias


def generate_key(
runner: Runner, key_path: Path, pub_path: Path, file_alias: str,
) -> None:
def generate_key(runner: Runner, key_path: Path) -> None:
"""Generate an Ed25519 key pair."""
if runner.mode == "native":
rc = runner.run([
"ssh-keygen", "-t", "ed25519", "-N", "", "-f", str(key_path),
])
else:
rc = runner.run(
f"ssh-keygen -t ed25519 -N '' -f ~/.ssh/id_ed25519_{file_alias}"
f"ssh-keygen -t ed25519 -N '' -f ~/.ssh/{key_path.name}"
)

if rc != 0:
Expand Down Expand Up @@ -657,10 +660,10 @@ def main() -> None:
if cli.ask_yn("Regenerate key pair?"):
key_path.unlink(missing_ok=True)
pub_path.unlink()
generate_key(runner, key_path, pub_path, file_alias)
generate_key(runner, key_path)
key_generated = True
else:
generate_key(runner, key_path, pub_path, file_alias)
generate_key(runner, key_path)
key_generated = True

cli.separator()
Expand Down