scrapli.transport.base.base_socket
Socket
Source code in transport/base/base_socket.py
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 | class Socket:
def __init__(self, host: str, port: int, timeout: float):
"""
Socket object
Args:
host: host to connect to
port: port to connect to
timeout: timeout in seconds
Returns:
None
Raises:
N/A
"""
self.logger = get_instance_logger(instance_name="scrapli.socket", host=host, port=port)
self.host = host
self.port = port
self.timeout = timeout
self.sock: Optional[socket.socket] = None
def __bool__(self) -> bool:
"""
Magic bool method for Socket
Args:
N/A
Returns:
bool: True/False if socket is alive or not
Raises:
N/A
"""
return self.isalive()
def _connect(self, socket_address_families: Set["socket.AddressFamily"]) -> None:
"""
Try to open socket to host using all possible address families
It seems that very occasionally when resolving a hostname (i.e. localhost during functional
tests against vrouter devices), a v6 address family will be the first af the socket
getaddrinfo returns, in this case, because the qemu hostfwd is not listening on ::1, instead
only listening on 127.0.0.1 the connection will fail. Presumably this is something that can
happen in real life too... something gets resolved with a v6 address but is denying
connections or just not listening on that ipv6 address. This little connect wrapper is
intended to deal with these weird scenarios.
Args:
socket_address_families: set of address families available for the provided host
really only should ever be v4 AND v6 if providing a hostname that resolves with
both addresses, otherwise if you just provide a v4/v6 address it will just be a
single address family for that type of address
Returns:
None
Raises:
ScrapliConnectionNotOpened: if socket refuses connection on all address families
ScrapliConnectionNotOpened: if socket connection times out on all address families
"""
for address_family_index, address_family in enumerate(socket_address_families, start=1):
self.sock = socket.socket(address_family, socket.SOCK_STREAM)
self.sock.settimeout(self.timeout)
try:
self.sock.connect((self.host, self.port))
except ConnectionRefusedError as exc:
msg = (
f"connection refused trying to open socket to {self.host} on port {self.port} "
f"for address family {address_family.name}"
)
self.logger.warning(msg)
if address_family_index == len(socket_address_families):
raise ScrapliConnectionNotOpened(msg) from exc
except socket.timeout as exc:
msg = (
f"timed out trying to open socket to {self.host} on port {self.port} for "
f"address family {address_family.name}"
)
self.logger.warning(msg)
if address_family_index == len(socket_address_families):
raise ScrapliConnectionNotOpened(msg) from exc
else:
return
def open(self) -> None:
"""
Open underlying socket
Args:
N/A
Returns:
None
Raises:
ScrapliConnectionNotOpened: if cant fetch socket addr info
"""
self.logger.debug(f"opening socket connection to '{self.host}' on port '{self.port}'")
socket_address_families = None
with suppress(socket.gaierror):
sock_info = socket.getaddrinfo(self.host, self.port)
if sock_info:
# get all possible address families for the provided host/port
# should only ever be two... one for v4 and one for v6... i think/hope?! :)?
socket_address_families = {sock[0] for sock in sock_info}
if not socket_address_families:
# this will likely need to be clearer just dont know what failure scenarios exist for
# this yet...
raise ScrapliConnectionNotOpened("failed to determine socket address family for host")
if not self.isalive():
self._connect(socket_address_families=socket_address_families)
self.logger.debug(
f"opened socket connection to '{self.host}' on port '{self.port}' successfully"
)
def close(self) -> None:
"""
Close socket
Args:
N/A
Returns:
None
Raises:
N/A
"""
self.logger.debug(f"closing socket connection to '{self.host}' on port '{self.port}'")
if self.isalive() and isinstance(self.sock, socket.socket):
self.sock.close()
self.logger.debug(
f"closed socket connection to '{self.host}' on port '{self.port}' successfully"
)
def isalive(self) -> bool:
"""
Check if socket is alive
Args:
N/A
Returns:
bool True/False if socket is alive
Raises:
N/A
"""
try:
if isinstance(self.sock, socket.socket):
self.sock.send(b"")
return True
except OSError:
self.logger.debug(f"Socket to host {self.host} is not alive")
return False
return False
|
__bool__() -> bool
Magic bool method for Socket
Returns:
| Name | Type |
Description |
bool |
bool
|
True/False if socket is alive or not |
Source code in transport/base/base_socket.py
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49 | def __bool__(self) -> bool:
"""
Magic bool method for Socket
Args:
N/A
Returns:
bool: True/False if socket is alive or not
Raises:
N/A
"""
return self.isalive()
|
__init__(host: str, port: int, timeout: float)
Socket object
Parameters:
| Name |
Type |
Description |
Default |
host |
str
|
host to connect to |
required
|
port |
int
|
port to connect to |
required
|
timeout |
float
|
timeout in seconds |
required
|
Returns:
Source code in transport/base/base_socket.py
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33 | def __init__(self, host: str, port: int, timeout: float):
"""
Socket object
Args:
host: host to connect to
port: port to connect to
timeout: timeout in seconds
Returns:
None
Raises:
N/A
"""
self.logger = get_instance_logger(instance_name="scrapli.socket", host=host, port=port)
self.host = host
self.port = port
self.timeout = timeout
self.sock: Optional[socket.socket] = None
|
close() -> None
Close socket
Returns:
| Type |
Description |
None
|
None |
Source code in transport/base/base_socket.py
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159 | def close(self) -> None:
"""
Close socket
Args:
N/A
Returns:
None
Raises:
N/A
"""
self.logger.debug(f"closing socket connection to '{self.host}' on port '{self.port}'")
if self.isalive() and isinstance(self.sock, socket.socket):
self.sock.close()
self.logger.debug(
f"closed socket connection to '{self.host}' on port '{self.port}' successfully"
)
|
isalive() -> bool
Check if socket is alive
Returns:
| Type |
Description |
bool
|
bool True/False if socket is alive |
Source code in transport/base/base_socket.py
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182 | def isalive(self) -> bool:
"""
Check if socket is alive
Args:
N/A
Returns:
bool True/False if socket is alive
Raises:
N/A
"""
try:
if isinstance(self.sock, socket.socket):
self.sock.send(b"")
return True
except OSError:
self.logger.debug(f"Socket to host {self.host} is not alive")
return False
return False
|
open() -> None
Open underlying socket
Returns:
| Type |
Description |
None
|
None |
Raises:
| Type |
Description |
ScrapliConnectionNotOpened
|
if cant fetch socket addr info |
Source code in transport/base/base_socket.py
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 | def open(self) -> None:
"""
Open underlying socket
Args:
N/A
Returns:
None
Raises:
ScrapliConnectionNotOpened: if cant fetch socket addr info
"""
self.logger.debug(f"opening socket connection to '{self.host}' on port '{self.port}'")
socket_address_families = None
with suppress(socket.gaierror):
sock_info = socket.getaddrinfo(self.host, self.port)
if sock_info:
# get all possible address families for the provided host/port
# should only ever be two... one for v4 and one for v6... i think/hope?! :)?
socket_address_families = {sock[0] for sock in sock_info}
if not socket_address_families:
# this will likely need to be clearer just dont know what failure scenarios exist for
# this yet...
raise ScrapliConnectionNotOpened("failed to determine socket address family for host")
if not self.isalive():
self._connect(socket_address_families=socket_address_families)
self.logger.debug(
f"opened socket connection to '{self.host}' on port '{self.port}' successfully"
)
|