remote.py 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164
  1. #!/usr/bin/env python3
  2. """SSH helper for the worker: key first, then WORKER_PASS from .env / env."""
  3. from __future__ import annotations
  4. import argparse
  5. import os
  6. import sys
  7. import pexpect
  8. def load_dotenv(path: str) -> None:
  9. if not os.path.isfile(path):
  10. return
  11. with open(path) as f:
  12. for line in f:
  13. line = line.strip()
  14. if not line or line.startswith("#") or "=" not in line:
  15. continue
  16. k, v = line.split("=", 1)
  17. v = v.strip().strip('"').strip("'")
  18. os.environ.setdefault(k.strip(), v)
  19. def ssh_cmd(
  20. host: str,
  21. user: str,
  22. remote: str,
  23. password: str | None,
  24. identity: str | None,
  25. timeout: int,
  26. ) -> int:
  27. dest = f"{user}@{host}"
  28. base = [
  29. "ssh",
  30. "-o",
  31. "StrictHostKeyChecking=accept-new",
  32. "-o",
  33. "ConnectTimeout=15",
  34. "-o",
  35. "ServerAliveInterval=15",
  36. ]
  37. # Prefer key when present
  38. if identity and os.path.isfile(os.path.expanduser(identity)):
  39. key = os.path.expanduser(identity)
  40. cmd = base + [
  41. "-o",
  42. "BatchMode=yes",
  43. "-o",
  44. "IdentitiesOnly=yes",
  45. "-i",
  46. key,
  47. dest,
  48. remote,
  49. ]
  50. # timeout=None (or <=0): wait forever — needed for docker pull / rsync.
  51. child = pexpect.spawn(
  52. cmd[0], cmd[1:], timeout=(None if timeout <= 0 else timeout), encoding="utf-8"
  53. )
  54. # Drain output until EOF; reset idle timer on each chunk so long pulls
  55. # with sparse progress lines don't die at a fixed wall clock.
  56. buf: list[str] = []
  57. while True:
  58. i = child.expect([pexpect.EOF, pexpect.TIMEOUT], timeout=(None if timeout <= 0 else timeout))
  59. if child.before:
  60. sys.stdout.write(child.before)
  61. sys.stdout.flush()
  62. buf.append(child.before)
  63. if i == 0:
  64. break
  65. # TIMEOUT with no EOF: still alive but quiet — keep waiting
  66. continue
  67. child.close()
  68. if child.exitstatus == 0:
  69. return 0
  70. # Key auth ran but remote command failed — don't fall through to password.
  71. if child.exitstatus is not None:
  72. return child.exitstatus
  73. if not password:
  74. print(
  75. "SSH key auth failed and WORKER_PASS is unset/empty.",
  76. file=sys.stderr,
  77. )
  78. return 255
  79. cmd = base + [
  80. "-o",
  81. "PreferredAuthentications=password",
  82. "-o",
  83. "PubkeyAuthentication=no",
  84. "-o",
  85. "NumberOfPasswordPrompts=1",
  86. dest,
  87. remote,
  88. ]
  89. to = None if timeout <= 0 else timeout
  90. child = pexpect.spawn(cmd[0], cmd[1:], timeout=to, encoding="utf-8")
  91. i = child.expect(
  92. [r"(?i)password:", r"(?i)permission denied", pexpect.EOF, pexpect.TIMEOUT]
  93. )
  94. if i != 0:
  95. print(
  96. "SSH password prompt not received "
  97. f"(got index={i}). Check WORKER_USER/WORKER_HOST and that "
  98. "password auth is enabled on the worker.",
  99. file=sys.stderr,
  100. )
  101. sys.stderr.write(child.before or "")
  102. child.close()
  103. return 255
  104. child.sendline(password)
  105. while True:
  106. j = child.expect([pexpect.EOF, pexpect.TIMEOUT], timeout=to)
  107. if child.before:
  108. sys.stdout.write(child.before)
  109. sys.stdout.flush()
  110. if j == 0:
  111. break
  112. child.close()
  113. if child.exitstatus not in (0, None):
  114. print(
  115. f"SSH to {dest} failed (exit {child.exitstatus}). "
  116. "Update WORKER_PASS in .env or install the shared SSH key on the worker.",
  117. file=sys.stderr,
  118. )
  119. return child.exitstatus or 255
  120. return 0
  121. def main() -> int:
  122. ap = argparse.ArgumentParser()
  123. ap.add_argument("--env-file", default="")
  124. ap.add_argument("--host", default=os.environ.get("WORKER_HOST", "10.0.0.2"))
  125. ap.add_argument("--user", default=os.environ.get("WORKER_USER") or os.environ.get("USER", "spark"))
  126. ap.add_argument(
  127. "--identity",
  128. default=os.environ.get("SSH_IDENTITY", "~/.ssh/id_ed25519_shared"),
  129. )
  130. # Seconds of idle with no output before abort. 0 = wait forever.
  131. # Default 600: short commands finish fast; docker pull needs a high value
  132. # (start.sh passes --timeout for long ops).
  133. ap.add_argument(
  134. "--timeout",
  135. type=int,
  136. default=int(os.environ.get("REMOTE_TIMEOUT", "600")),
  137. )
  138. ap.add_argument("remote")
  139. args = ap.parse_args()
  140. if args.env_file:
  141. load_dotenv(args.env_file)
  142. password = os.environ.get("WORKER_PASS") or None
  143. return ssh_cmd(
  144. args.host,
  145. args.user,
  146. args.remote,
  147. password,
  148. args.identity,
  149. args.timeout,
  150. )
  151. if __name__ == "__main__":
  152. raise SystemExit(main())