Skip to content

Session Manager

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.

Required IAM Permissions:

  • ec2:DescribeInstances
  • ssm:StartSession
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
@session_manager_app.command()
def port_forward(  # noqa: PLR0913
    *,
    # 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.

    Required IAM Permissions:

    - `ec2:DescribeInstances`
    - `ssm:StartSession`
    """
    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,
    )
    proc = subprocess.Popen(  # noqa: S603
        command,
        stdout=stdout,
        stderr=subprocess.STDOUT,
        text=True,
        close_fds=False,  # FD inherited from parent process
    )
    pid = proc.pid

    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
    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.

Required IAM Permissions:

  • ec2:DescribeInstances
  • ssm:StartSession
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
64
65
@session_manager_app.command()
def start(
    *,
    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.

    Required IAM Permissions:

    - `ec2:DescribeInstances`
    - `ssm:StartSession`
    """
    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,
    )
    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
@session_manager_app.command()
def stop(
    *,
    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."""
    # 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)
        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)
        pid_file.unlink()

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