Skip to content

Session Manager

aws-annoying session-manager install

Install AWS Session Manager plugin.

Source code in aws_annoying/cli/session_manager/install.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@session_manager_app.command()
def install(
    ctx: typer.Context,
    *,
    yes: bool = typer.Option(
        False,  # noqa: FBT003
        help="Do not ask confirmation for installation.",
    ),
) -> None:
    """Install AWS Session Manager plugin."""
    dry_run = ctx.meta["dry_run"]
    session_manager = SessionManager()

    # Check session-manager-plugin already installed
    is_installed, binary_path, version = session_manager.verify_installation()
    if is_installed:
        logger.info("Session Manager plugin is already installed at %s (version: %s)", binary_path, version)
        return

    # Install session-manager-plugin
    logger.warning("Installing AWS Session Manager plugin. You could be prompted for admin privileges request.")
    if not dry_run:
        session_manager.install(confirm=yes, downloader=TQDMDownloader())

    # Verify installation
    is_installed, binary_path, version = session_manager.verify_installation()
    if not is_installed:
        logger.error("Installation failed. Session Manager plugin not found.")
        raise typer.Exit(1)

    logger.info("Session Manager plugin successfully installed at %s (version: %s)", binary_path, version)

aws-annoying session-manager port-forward

Start a port forwarding session using AWS Session Manager.

This command allows starting a port forwarding session through an EC2 instance identified by its name or ID. If there are more than one instance with the same name, the first one found will be used.

Also, it manages a PID file to keep track of the session manager plugin process running in background, allowing to terminate any existing process before starting a new one.

Source code in aws_annoying/cli/session_manager/port_forward.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@session_manager_app.command()
def port_forward(  # noqa: PLR0913
    ctx: typer.Context,
    *,
    # TODO(lasuillard): Add `--local-host` option, redirect the traffic to non-localhost bind (unsupported by AWS)
    local_port: int = typer.Option(
        ...,
        show_default=False,
        help="The local port to use for port forwarding.",
    ),
    through: str = typer.Option(
        ...,
        show_default=False,
        help="The name or ID of the EC2 instance to use as a proxy for port forwarding.",
    ),
    remote_host: str = typer.Option(
        ...,
        show_default=False,
        help="The remote host to connect to.",
    ),
    remote_port: int = typer.Option(
        ...,
        show_default=False,
        help="The remote port to connect to.",
    ),
    reason: str = typer.Option(
        "",
        help="The reason for starting the port forwarding session.",
    ),
    pid_file: Path = typer.Option(  # noqa: B008
        "./session-manager-plugin.pid",
        help="The path to the PID file to store the process ID of the session manager plugin.",
    ),
    terminate_running_process: bool = typer.Option(
        False,  # noqa: FBT003
        help="Terminate the process in the PID file if it already exists.",
    ),
    log_file: Path = typer.Option(  # noqa: B008
        "./session-manager-plugin.log",
        help="The path to the log file to store the output of the session manager plugin.",
    ),
) -> None:
    """Start a port forwarding session using AWS Session Manager.

    This command allows starting a port forwarding session through an EC2 instance identified by its name or ID.
    If there are more than one instance with the same name, the first one found will be used.

    Also, it manages a PID file to keep track of the session manager plugin process running in background,
    allowing to terminate any existing process before starting a new one.
    """
    dry_run = ctx.meta["dry_run"]
    session_manager = SessionManager()

    # Check if the PID file already exists
    if pid_file.exists():
        if not terminate_running_process:
            logger.error("PID file already exists.")
            raise typer.Exit(1)

        pid_content = pid_file.read_text()
        try:
            existing_pid = int(pid_content)
        except ValueError:
            logger.error("PID file content is invalid; expected integer, but got: %r", type(pid_content))  # noqa: TRY400
            raise typer.Exit(1) from None

        try:
            logger.warning("Terminating running process with PID %d.", existing_pid)
            os.kill(existing_pid, signal.SIGTERM)
            pid_file.write_text("")  # Clear the PID file
        except ProcessLookupError:
            logger.warning("Tried to terminate process with PID %d but does not exist.", existing_pid)

    # Resolve the instance name or ID
    instance_id = get_instance_id_by_name(through)
    if instance_id:
        logger.info("Instance ID resolved: [bold]%s[/bold]", instance_id)
        target = instance_id
    else:
        logger.error("Instance with name '%s' not found.", through)
        raise typer.Exit(1)

    # Initiate the session
    command = session_manager.build_command(
        target=target,
        document_name="AWS-StartPortForwardingSessionToRemoteHost",
        parameters={
            "host": [remote_host],
            "portNumber": [str(remote_port)],
            "localPortNumber": [str(local_port)],
        },
        reason=reason,
    )
    stdout: subprocess._FILE
    if log_file is not None:  # noqa: SIM108
        stdout = log_file.open(mode="at+", buffering=1)
    else:
        stdout = subprocess.DEVNULL

    logger.info(
        "Starting port forwarding session through [bold]%s[/bold] with reason: [italic]%r[/italic].",
        through,
        reason,
    )
    if not dry_run:
        proc = subprocess.Popen(  # noqa: S603
            command,
            stdout=stdout,
            stderr=subprocess.STDOUT,
            text=True,
            close_fds=False,  # FD inherited from parent process
        )
        pid = proc.pid
    else:
        pid = -1

    logger.info(
        "Session Manager Plugin started with PID %d. Outputs will be logged to %s.",
        pid,
        log_file.absolute(),
    )

    # Write the PID to the file
    if not dry_run:
        pid_file.write_text(str(pid))

    logger.info("PID file written to %s.", pid_file.absolute())

aws-annoying session-manager start

Start new session to your instance.

You can use your EC2 instance identified by its name or ID. If there are more than one instance with the same name, the first one found will be used.

Source code in aws_annoying/cli/session_manager/start.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@session_manager_app.command()
def start(
    ctx: typer.Context,
    *,
    target: str = typer.Option(
        ...,
        show_default=False,
        help="The name or ID of the EC2 instance to connect to.",
    ),
    reason: str = typer.Option(
        "",
        help="The reason for starting the session.",
    ),
) -> None:
    """Start new session to your instance.

    You can use your EC2 instance identified by its name or ID. If there are
    more than one instance with the same name, the first one found will be used.
    """
    dry_run = ctx.meta["dry_run"]
    session_manager = SessionManager()

    # Resolve the instance name or ID
    instance_id = get_instance_id_by_name(target)
    if instance_id:
        logger.info("Instance ID resolved: [bold]%s[/bold]", instance_id)
        target = instance_id
    else:
        logger.info("Instance with name '%s' not found.", target)
        raise typer.Exit(1)

    # Start the session, replacing the current process
    logger.info(
        "Starting session to target [bold]%s[/bold] with reason: [italic]%r[/italic].",
        target,
        reason,
    )
    command = session_manager.build_command(
        target=target,
        document_name="SSM-SessionManagerRunShell",
        parameters={},
        reason=reason,
    )
    if not dry_run:
        os.execvp(command[0], command)  # noqa: S606

aws-annoying session-manager stop

Stop running session for PID file.

Source code in aws_annoying/cli/session_manager/stop.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@session_manager_app.command()
def stop(
    ctx: typer.Context,
    *,
    pid_file: Path = typer.Option(  # noqa: B008
        "./session-manager-plugin.pid",
        help="The path to the PID file to store the process ID of the session manager plugin.",
    ),
    remove: bool = typer.Option(
        True,  # noqa: FBT003
        help="Remove the PID file after stopping the session.",
    ),
) -> None:
    """Stop running session for PID file."""
    dry_run = ctx.meta["dry_run"]

    # Check if PID file exists
    if not pid_file.is_file():
        logger.error("PID file not found: %s", pid_file)
        raise typer.Exit(1)

    # Read PID from file
    pid_content = pid_file.read_text()
    try:
        pid = int(pid_content)
    except ValueError:
        logger.error("PID file content is invalid; expected integer, but got: %s", type(pid_content))  # noqa: TRY400
        raise typer.Exit(1) from None

    # Send SIGTERM to the process
    try:
        logger.warning("Terminating running process with PID %d.", pid)
        if not dry_run:
            os.kill(pid, signal.SIGTERM)
    except ProcessLookupError:
        logger.warning("Tried to terminate process with PID %d but does not exist.", pid)

    # Remove the PID file
    if remove:
        logger.info("Removed the PID file %s.", pid_file)
        if not dry_run:
            pid_file.unlink()

    logger.info("Terminated the session successfully.")