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.

Required IAM Permissions:

  • ec2:DescribeInstances
  • ssm:StartSession
Source code in aws_annoying/_cli/session_manager/port_forward.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
 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
@session_manager_app.command()
def port_forward(  # noqa: PLR0913
    *,
    local_host: str = typer.Option(
        "127.0.0.1",
        help="The local host address to bind the port forwarding connection.",
    ),
    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.",
    ),
) -> 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.

    Required IAM Permissions:

    - `ec2:DescribeInstances`
    - `ssm:StartSession`
    """
    session_manager = SessionManager()

    # 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)

    is_non_localhost = local_host not in ("127.0.0.1", "localhost")
    ssm_local_host = "127.0.0.1"
    ssm_local_port = get_free_port() if is_non_localhost else local_port

    # Initiate the session
    command = session_manager.build_command(
        target=target,
        document_name="AWS-StartPortForwardingSessionToRemoteHost",
        parameters={
            "host": [remote_host],
            "portNumber": [str(remote_port)],
            "localPortNumber": [str(ssm_local_port)],
        },
        reason=reason,
    )

    logger.info(
        "Starting port forwarding session through [bold]%s[/bold] with reason: [italic]%r[/italic].",
        through,
        reason,
    )

    proxy: TCPProxy | None = None
    if is_non_localhost:
        proxy = TCPProxy(Address(local_host, local_port), Address(ssm_local_host, ssm_local_port))
        proxy.start()
        logger.info(
            "TCP Proxy started on %s:%d -> %s:%d",
            local_host,
            local_port,
            ssm_local_host,
            ssm_local_port,
        )

    # NOTE: stdout=None inherits the parent process's standard output stream without
    #       requiring a concrete file descriptor (sys.stdout.fileno()), preventing errors in
    #       environments where stdout is captured (e.g. pytest or CLI test runners).
    proc = subprocess.Popen(  # noqa: S603
        command,
        stdout=None,
        stderr=subprocess.STDOUT,
        text=True,
    )

    def handle_signal(sig: int, _frame: Any) -> None:
        logger.info("Received signal %d, stopping port forwarding...", sig)
        proc.terminate()
        if proxy is not None:
            proxy.stop()

    signal.signal(signal.SIGINT, handle_signal)
    signal.signal(signal.SIGTERM, handle_signal)

    try:
        return_code = proc.wait()
    finally:
        if proxy is not None:
            proxy.stop()

    if return_code != 0:
        raise typer.Exit(return_code)

aws-annoying session-manager start

Start new session to your instance or ECS container.

You can use your EC2 instance identified by its name or ID, or an ECS target formatted as ecs:<cluster-name>_<task-id>_<container-runtime-id>.

Required IAM Permissions:

  • For EC2 instances:
  • ec2:DescribeInstances
  • ssm:StartSession
  • For ECS Exec (containers):
  • ecs:DescribeTasks
  • ecs:ListClusters
  • ecs:ListServices
  • ecs:ListTasks
  • ssm:StartSession (uses the SSM Session Manager instead of ECS Exec directly)
Source code in aws_annoying/_cli/session_manager/start.py 
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
@session_manager_app.command()
def start(
    *,
    target: str | None = typer.Option(
        None,
        help="The name or ID of the EC2 instance, or ECS target to connect to. If omitted, prompts interactively.",
    ),
    reason: str = typer.Option(
        "",
        help="The reason for starting the session.",
    ),
) -> None:
    """Start new session to your instance or ECS container.

    You can use your EC2 instance identified by its name or ID, or an ECS target
    formatted as `ecs:<cluster-name>_<task-id>_<container-runtime-id>`.

    Required IAM Permissions:

    - For EC2 instances:
      - `ec2:DescribeInstances`
      - `ssm:StartSession`
    - For ECS Exec (containers):
      - `ecs:DescribeTasks`
      - `ecs:ListClusters`
      - `ecs:ListServices`
      - `ecs:ListTasks`
      - `ssm:StartSession` (uses the SSM Session Manager instead of ECS Exec directly)
    """
    session_manager = SessionManager()

    if target is None:
        target = _handle_interactive_start()

    if not _is_ecs_target(target):
        # EC2 logic: 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