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
247
248
249
250
|
# 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:
# Sam Crooks - initial implementation
# Ron Frederick - minor cleanup
"""Logging functions"""
import logging
from typing import MutableMapping, Optional, Tuple, Union, cast
_LogArg = object
_ObjDict = MutableMapping[str, object]
class SSHLogger(logging.LoggerAdapter):
"""Adapter to add context to AsyncSSH log messages"""
_debug_level = 1
_pkg_logger = logging.getLogger(__package__ or 'asyncssh')
def __init__(self, parent: logging.Logger = _pkg_logger,
child: str = '', context: str = ''):
self._context = context
self._logger = parent.getChild(child) if child else parent
super().__init__(self._logger, {})
def _extend_context(self, context: str) -> str:
"""Extend context provided by this logger"""
if context:
if self._context:
context = self._context + ', ' + context
else:
context = self._context
return context
def get_child(self, child: str = '', context: str = '') -> 'SSHLogger':
"""Return child logger with optional added context"""
return type(self)(self._logger, child, self._extend_context(context))
def log(self, level: int, msg: object, *args, **kwargs) -> None:
"""Log a message to the underlying logger"""
def _item_text(item: _LogArg) -> str:
"""Convert a list item to text"""
if isinstance(item, bytes):
result = item.decode('utf-8', errors='backslashreplace')
if not result.isprintable():
result = repr(result)[1:-1]
elif not isinstance(item, str):
result = str(item)
else:
result = item
return result
def _text(arg: _LogArg) -> _LogArg:
"""Convert a log argument to text"""
result: _LogArg
if isinstance(arg, list):
result = ','.join(_item_text(item) for item in arg)
elif isinstance(arg, tuple):
host, port = arg
if host:
result = f'{host}, port {port}' if port else host
else:
result = f'port {port}' if port else 'dynamic port'
elif isinstance(arg, bytes):
result = _item_text(arg)
else:
result = arg
return result
log_args = [_text(arg) for arg in args]
super().log(level, msg, *log_args, **kwargs)
def process(self, msg: str, kwargs: _ObjDict) -> Tuple[str, _ObjDict]:
"""Add context to log message"""
extra = cast(_ObjDict, kwargs.get('extra', {}))
context = self._extend_context(cast(str, extra.get('context')))
context = '[' + context + '] ' if context else ''
packet = cast(bytes, extra.get('packet'))
pktdata = ''
offset = 0
while packet:
line = f'\n {offset:08x}:'
for b in packet[:16]:
line += f' {b:02x}'
line += (62 - len(line)) * ' '
for b in packet[:16]:
if b < 0x20 or b >= 0x80:
c = '.'
elif b == ord('%'):
c = '%%'
else:
c = chr(b)
line += c
pktdata += line
packet = packet[16:]
offset += 16
return context + msg + pktdata, kwargs
@classmethod
def set_debug_level(cls, level: int) -> None:
"""Set AsyncSSH debug log level"""
if level < 1 or level > 3:
raise ValueError('Debug log level must be between 1 and 3')
cls._debug_level = level
def debug1(self, msg: str, *args: _LogArg, **kwargs: object) -> None:
"""Write a level 1 debug log message"""
self.log(logging.DEBUG, msg, *args, **kwargs)
def debug2(self, msg: str, *args: _LogArg, **kwargs: object) -> None:
"""Write a level 2 debug log message"""
if self._debug_level >= 2:
self.log(logging.DEBUG, msg, *args, **kwargs)
def packet(self, pktid: Optional[int], packet: bytes, msg: str,
*args: _LogArg, **kwargs: object) -> None:
"""Write a control packet debug log message"""
if self._debug_level >= 3:
kwargs.setdefault('extra', {})
extra = cast(_ObjDict, kwargs.get('extra'))
if pktid is not None:
extra.update(context=f'pktid={pktid}')
extra.update(packet=packet)
self.log(logging.DEBUG, msg, *args, **kwargs)
def set_log_level(level: Union[int, str]) -> None:
"""Set the AsyncSSH log level
This function sets the log level of the AsyncSSH logger. It
defaults to `'NOTSET`', meaning that it will track the debug
level set on the root Python logger.
For additional control over the level of debug logging, see the
function :func:`set_debug_level` for additional information.
:param level:
The log level to set, as defined by the `logging` module
:type level: `int` or `str`
"""
logger.setLevel(level)
def set_sftp_log_level(level: Union[int, str]) -> None:
"""Set the AsyncSSH SFTP/SCP log level
This function sets the log level of the AsyncSSH SFTP/SCP logger.
It defaults to `'NOTSET`', meaning that it will track the debug
level set on the main AsyncSSH logger.
For additional control over the level of debug logging, see the
function :func:`set_debug_level` for additional information.
:param level:
The log level to set, as defined by the `logging` module
:type level: `int` or `str`
"""
sftp_logger.setLevel(level)
def set_debug_level(level: int) -> None:
"""Set the AsyncSSH debug log level
This function sets the level of debugging logging done by the
AsyncSSH logger, from the following options:
===== ====================================
Level Description
===== ====================================
1 Minimal debug logging
2 Full debug logging
3 Full debug logging with packet dumps
===== ====================================
The debug level defaults to level 1 (minimal debug logging).
.. note:: For this setting to have any effect, the effective log
level of the AsyncSSH logger must be set to DEBUG.
.. warning:: Extreme caution should be used when setting debug
level to 3, as this can expose user passwords in
clear text. This level should generally only be
needed when tracking down issues with malformed
or incomplete packets.
:param level:
The debug level to set, as defined above.
:type level: `int`
"""
logger.set_debug_level(level)
logger = SSHLogger()
sftp_logger = logger.get_child('sftp')
|
