Runs a command over SSH via EC2 Instance Connect.
(
instance: Instance,
command: str,
*,
extra_ssh_args: list[str] | None = None,
input: bytes | None = None,
quiet: bool = False,
)
| 227 | |
| 228 | |
| 229 | def mssh( |
| 230 | instance: Instance, |
| 231 | command: str, |
| 232 | *, |
| 233 | extra_ssh_args: list[str] | None = None, |
| 234 | input: bytes | None = None, |
| 235 | quiet: bool = False, |
| 236 | ) -> None: |
| 237 | """Runs a command over SSH via EC2 Instance Connect.""" |
| 238 | extra_ssh_args = extra_ssh_args or [] |
| 239 | host = instance_host(instance) |
| 240 | if command: |
| 241 | if not quiet: |
| 242 | print(f"{host}$ {command}", file=sys.stderr) |
| 243 | # Quote to work around: |
| 244 | # https://github.com/aws/aws-ec2-instance-connect-cli/pull/26 |
| 245 | command = shlex.quote(command) |
| 246 | else: |
| 247 | print(f"$ mssh {host}") |
| 248 | |
| 249 | result = subprocess.run( |
| 250 | [ |
| 251 | *SSH_COMMAND, |
| 252 | *extra_ssh_args, |
| 253 | host, |
| 254 | command, |
| 255 | ], |
| 256 | input=input, |
| 257 | stdout=subprocess.DEVNULL if quiet else None, |
| 258 | stderr=subprocess.DEVNULL if quiet else None, |
| 259 | ) |
| 260 | # Exit code 130 = SIGINT (Ctrl-C) — exit cleanly |
| 261 | if result.returncode == 130: |
| 262 | sys.exit(130) |
| 263 | if result.returncode != 0: |
| 264 | raise subprocess.CalledProcessError(result.returncode, result.args) |
| 265 | |
| 266 | |
| 267 | def msftp( |
no test coverage detected