1
2
3
4
5
6
7
8
9
10
11
12
13
14
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
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
|
# Copyright (c) 2013-2024 by Ron Frederick <ronf@timeheart.net> and others.
#
# This program and the accompanying materials are made available under
# the terms of the Eclipse Public License v2.0 which accompanies this
# distribution and is available at:
#
# http://www.eclipse.org/legal/epl-2.0/
#
# This program may also be made available under the following secondary
# licenses when the conditions for such availability set forth in the
# Eclipse Public License v2.0 are satisfied:
#
# GNU General Public License, Version 2.0, or any later versions of
# that license
#
# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
#
# Contributors:
# Ron Frederick - initial implementation, API, and documentation
"""SSH port forwarding handlers"""
import asyncio
import socket
from types import TracebackType
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Optional
from typing import Type, cast
from typing_extensions import Self
from .misc import ChannelOpenError, SockAddr
if TYPE_CHECKING:
# pylint: disable=cyclic-import
from .connection import SSHConnection
SSHForwarderCoro = Callable[..., Awaitable]
class SSHForwarder(asyncio.BaseProtocol):
"""SSH port forwarding connection handler"""
def __init__(self, peer: Optional['SSHForwarder'] = None,
extra: Optional[Dict[str, Any]] = None):
self._peer = peer
self._transport: Optional[asyncio.Transport] = None
self._inpbuf = b''
self._eof_received = False
if peer:
peer.set_peer(self)
if extra is None:
extra = {}
self._extra = extra
async def __aenter__(self) -> Self:
return self
async def __aexit__(self, _exc_type: Optional[Type[BaseException]],
_exc_value: Optional[BaseException],
_traceback: Optional[TracebackType]) -> bool:
self.close()
return False
def get_extra_info(self, name: str, default: Any = None) -> Any:
"""Get additional information about the forwarder
This method returns extra information about the forwarder.
Currently, the only information available is the value
``interface`` for TUN/TAP forwarders, returning the name of the
local TUN/TAP network interface created for this forwarder.
"""
return self._extra.get(name, default)
def set_peer(self, peer: 'SSHForwarder') -> None:
"""Set the peer forwarder to exchange data with"""
self._peer = peer
def write(self, data: bytes) -> None:
"""Write data to the transport"""
assert self._transport is not None
self._transport.write(data)
def write_eof(self) -> None:
"""Write end of file to the transport"""
assert self._transport is not None
try:
self._transport.write_eof()
except OSError: # pragma: no cover
pass
def was_eof_received(self) -> bool:
"""Return whether end of file has been received or not"""
return self._eof_received
def pause_reading(self) -> None:
"""Pause reading from the transport"""
assert self._transport is not None
self._transport.pause_reading()
def resume_reading(self) -> None:
"""Resume reading on the transport"""
assert self._transport is not None
self._transport.resume_reading()
def connection_made(self, transport: asyncio.BaseTransport) -> None:
"""Handle a newly opened connection"""
self._transport = cast(Optional['asyncio.Transport'], transport)
sock = cast(socket.socket, transport.get_extra_info('socket'))
if sock and sock.family in {socket.AF_INET, socket.AF_INET6}:
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
def connection_lost(self, exc: Optional[Exception]) -> None:
"""Handle an incoming connection close"""
# pylint: disable=unused-argument
self.close()
def session_started(self) -> None:
"""Handle session start"""
def data_received(self, data: bytes,
datatype: Optional[int] = None) -> None:
"""Handle incoming data from the transport"""
# pylint: disable=unused-argument
if self._peer:
self._peer.write(data)
else:
self._inpbuf += data
def eof_received(self) -> bool:
"""Handle an incoming end of file from the transport"""
self._eof_received = True
if self._peer:
self._peer.write_eof()
return not self._peer.was_eof_received()
else:
return True
def pause_writing(self) -> None:
"""Pause writing by asking peer to pause reading"""
if self._peer: # pragma: no branch
self._peer.pause_reading()
def resume_writing(self) -> None:
"""Resume writing by asking peer to resume reading"""
if self._peer: # pragma: no branch
self._peer.resume_reading()
def close(self) -> None:
"""Close this port forwarder"""
if self._transport:
self._transport.close()
self._transport = None
if self._peer:
peer = self._peer
self._peer = None
peer.close()
class SSHLocalForwarder(SSHForwarder):
"""Local forwarding connection handler"""
def __init__(self, conn: 'SSHConnection', coro: SSHForwarderCoro):
super().__init__()
self._conn = conn
self._coro = coro
async def _forward(self, *args: object) -> None:
"""Begin local forwarding"""
def session_factory() -> SSHForwarder:
"""Return an SSH forwarder"""
return SSHForwarder(self)
try:
await self._coro(session_factory, *args)
except ChannelOpenError as exc:
self.connection_lost(exc)
return
assert self._peer is not None
if self._inpbuf:
self._peer.write(self._inpbuf)
self._inpbuf = b''
if self._eof_received:
self._peer.write_eof()
def forward(self, *args: object) -> None:
"""Start a task to begin local forwarding"""
self._conn.create_task(self._forward(*args))
class SSHLocalPortForwarder(SSHLocalForwarder):
"""Local TCP port forwarding connection handler"""
def connection_made(self, transport: asyncio.BaseTransport) -> None:
"""Handle a newly opened connection"""
super().connection_made(transport)
peername = cast(SockAddr, transport.get_extra_info('peername'))
if peername: # pragma: no branch
orig_host, orig_port = peername[:2]
self.forward(orig_host, orig_port)
class SSHLocalPathForwarder(SSHLocalForwarder):
"""Local UNIX domain socket forwarding connection handler"""
def connection_made(self, transport: asyncio.BaseTransport) -> None:
"""Handle a newly opened connection"""
super().connection_made(transport)
self.forward()
|
