-
Notifications
You must be signed in to change notification settings - Fork 10
/
Copy pathikp3db.py
2103 lines (1824 loc) · 85.7 KB
/
ikp3db.py
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
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#! /usr/bin/env python
# coding: utf8
#
# This file is part of the IKP3db Debugger
# Copyright (c) 2016-2018 by Cyril MORISSE, Audaxis
# Licence: MIT. See LICENCE at repository root
#
import socket
import sys
import os
import atexit
import signal
import json
import logging
import traceback
import types
import inspect
import threading
import queue
import types
import argparse
import datetime
import io
import ctypes
import cgi
import iksettrace3
# For now ikpdb is a singleton
ikpdb = None
__version__ = "1.4.2"
##
# Logging System
# IKP3db has it's own logging system distinct from python logging to
# avoid collision when debugging programs which reconfigure logging
# system wide.
#
# logging is organized in domains (which corresponds to loggers)
# identified by one letter.
# IKP3db logs on these domains:
# letter: domain
# - n,N: Network
# - b,B: Breakpoints
# - e,E: Expression evaluation
# - x,X: Execution
# - f,F: Frame
# - g,G: Global debugger
#
# Logging support the same notion of level as python logging.
# Logging is invoked using this syntax:
# _logger.{{domain}}_{{level}}(*args)
# eg: _logger.x_debug("error in %s", the_error)
#
class ANSIColors(object):
MAGENTA = '\033[95m'
BLUE = '\033[94m' # debug
GREEN = '\033[92m' # info
YELLOW = '\033[93m' # warning
RED = '\033[91m' # error
BOLD = '\033[1m' # critical
UNDERLINE = '\033[4m'
ENDC = '\033[0m'
class IKPdbLoggerError(Exception):
pass
class MetaIKPdbLogger(type):
def __getattr__(cls, name):
domain, level_name = name.split('_')
level = IKPdbLogger.LEVELS.get(level_name, None)
if domain not in IKPdbLogger.DOMAINS or not level:
raise IKPdbLoggerError("'%s' is not valid logging domain and level combination !" % name)
def wrapper(*args, **kwargs):
return cls._log(domain, level, *args, **kwargs)
return wrapper
class IKPdbLogger(object, metaclass=MetaIKPdbLogger):
""" IKP3db implements it's own logging system to:
- avoid problem while debugging programs that reconfigure logging system wide.
- allow IKP3db debugging...
"""
enabled = False
TEMPLATES = [
"\033[1m[IKP3db-%s]\033[0m %s - \033[94mNOLOG\033[0m - %s", # nolog 0
"\033[1m[IKP3db-%s]\033[0m %s - \033[94mDEBUG\033[0m - %s", # debug 1
"\033[1m[IKP3db-%s]\033[0m %s - \033[92mINFO\033[0m - %s", # info 2
"\033[1m[IKP3db-%s]\033[0m %s - \033[93mWARNING\033[0m - %s", # warning 3
"\033[1m[IKP3db-%s]\033[0m %s - \033[91mERROR\033[0m - %s", # error 4
"\033[1m[IKP3db-%s]\033[0m %s - \033[91mCRITICAL\033[0m - %s", # critical 5
]
# Levels
CRITICAL = 50
ERROR = 40
WARNING= 30
INFO = 20
DEBUG = 10
NOLOG= 0
# Levels by name
LEVELS = {
"critical": 50,
"error": 40,
"warning": 30,
"info": 20,
"debug": 10,
"nolog": 0,
}
# Domains and domain's level
DOMAINS = {
"n": 20,
"b": 20,
"e": 20,
"x": 20,
"f": 20,
"p": 20,
"g": 20
}
@classmethod
def setup(cls, ikpdb_log_arg):
""" activates DEBUG logging level based on the `ikpdb_log_arg`
parameter string.
`ikpdb_log_arg` corresponds to the `--ikpdb-log` command line argument.
`ikpdb_log_arg` is composed of a serie of letters that set the `DEBUG`
logging level on the components of the debugger.
Here are the letters and the component they activate `DEBUG` logging
level on:
- n,N: Network
- b,B: Breakpoints
- e,E: Expression evaluation
- x,X: Execution
- f,F: Frame
- p,P: Path and python path manipulation
- g,G: Global debugger
By default logging is disabled for all components.
Any `ikpdb_log_arg` value different from the letters above (eg: '9')
activates `INFO` level logging on all domains.
To log, use::
_logger.x_debug("useful information")
Where:
- `_logger` is a reference to the IKPdbLogger class
- `x` is the `Execution` domain
- `debug` is the logging level
"""
if not ikpdb_log_arg:
return
IKPdbLogger.enabled = True
logging_configuration_string = ikpdb_log_arg.lower()
for letter in logging_configuration_string:
if letter in IKPdbLogger.DOMAINS:
IKPdbLogger.DOMAINS[letter] = 10
@classmethod
def _log(cls, domain, level, message, *args):
ts = datetime.datetime.now().strftime('%H:%M:%S,%f')
if level >= IKPdbLogger.DOMAINS[domain]:
try:
string = message % args
except:
string = message+"".join([str(e) for e in args])
print(IKPdbLogger.TEMPLATES[level//10] % (domain, ts, string,), file=sys.stderr, flush=True)
_logger = IKPdbLogger
##
# Network Manager
#
class IKPdbConnectionError(Exception):
pass
class IKPdbConnectionHandler(object):
""" IKPdbConnectionHandler manages a connection with a remote client once
it is established.
IKpdb and remote client exchanges messages having this structure:
``length={{integer length of json_message_body below}}{{MAGIC_CODE}}{{json_dump_of_message_body}}``
This class contains methods to receive, send and reply to such messages.
"""
MAGIC_CODE = u"LLADpcdtbdpac"
MESSAGE_TEMPLATE = u"length=%%i%s%%s" % MAGIC_CODE
SOCKET_BUFFER_SIZE = 4096 # Maximum size of a packet received from client
MSG_WAITALL = 0x100 # From Linux sys/socket.h
CMD_LOOP_SOCKET_TIMEOUT = 0.3 # second
def __init__(self, connection):
self._connection = connection
self._connection_lock = threading.Lock()
self._received_data = u''
self._network_loop = True
self._connection.settimeout(self.CMD_LOOP_SOCKET_TIMEOUT)
def encode(self, obj):
obj_str = json.dumps(obj)
msg_str = self.MESSAGE_TEMPLATE % (len(obj_str), obj_str,)
return msg_str
def decode(self, message):
json_obj = message.split(self.MAGIC_CODE)[1]
obj = json.loads(json_obj)
return obj
def log_sent(self, msg):
_logger.n_debug("Sent %s bytes >>>%s<<<", len(msg), msg)
def log_received(self, msg):
_logger.n_debug("Received %s bytes >>>%s<<<", len(msg), msg)
def send(self, command, _id=None, result={}, frames=[], threads=None,
error_messages=[], warning_messages=[], info_messages=[],
exception=None):
""" Build a message from parameters and send it to debugger.
:param command: The command sent to the debugger client.
:type command: str
:param _id: Unique id of the sent message. Right now, it's always `None`
for messages by debugger to client.
:type _id: int
:param result: Used to send `exit_code` and updated `executionStatus`
to debugger client.
:type result: dict
:param frames: contains the complete stack frames when debugger sends
the `programBreak` message.
:type frames: list
:param error_messages: A list of error messages the debugger client must
display to the user.
:type error_messages: list of str
:param warning_messages: A list of warning messages the debugger client
must display to the user.
:type warning_messages: list of str
:param info_messages: A list of info messages the debugger client must
display to the user.
:type info_messages: list of str
:param exception: If debugger encounter an exception, this dict contains
2 keys: `type` and `info` (the later is the message).
:type exception: dict
"""
with self._connection_lock:
payload = {
'_id': _id,
'command': command,
'result': result,
'commandExecStatus': 'ok',
'frames': frames,
'info_messages': info_messages,
'warning_messages': warning_messages,
'error_messages': error_messages,
'exception': exception
}
if threads:
payload['threads'] = threads
msg = self.encode(payload)
if self._connection:
msg_bytes = bytearray(msg, 'utf-8')
send_bytes_count = self._connection.sendall(msg_bytes)
self.log_sent(msg)
return send_bytes_count
raise IKPdbConnectionError("Connection lost!")
def reply(self, obj, result, command_exec_status='ok', info_messages=[],
warning_messages=[], error_messages=[]):
"""Build a response from a previouslsy received command message, send it
and return number of sent bytes.
:param result: Used to send back the result of the command execution to
the debugger client.
:type result: dict
See send() above for others parameters definition.
"""
with self._connection_lock:
# TODO: add a parameter to remove args from messages ?
if True:
del obj['args']
obj['result'] = result
obj['commandExecStatus'] = command_exec_status
obj['info_messages'] = info_messages
obj['warning_messages'] = warning_messages
obj['error_messages'] = error_messages
msg_str = self.encode(obj)
msg_bytes = bytearray(msg_str, 'utf-8')
send_bytes_count = self._connection.sendall(msg_bytes)
self.log_sent(msg_bytes)
return send_bytes_count
def receive(self, ikpdb):
"""Waits for a message from the debugger and returns it as a dict.
"""
# with self._connection_lock:
while self._network_loop:
_logger.n_debug("Enter socket.recv(%s) with self._received_data = %s",
self.SOCKET_BUFFER_SIZE,
self._received_data)
try:
# We may land here with a full packet already in self.received_data
# In that case we must not enter recv()
if self.SOCKET_BUFFER_SIZE:
data = self._connection.recv(self.SOCKET_BUFFER_SIZE)
else:
data = b''
_logger.n_debug("Socket.recv(%s) => %s", self.SOCKET_BUFFER_SIZE, data)
except socket.timeout:
_logger.n_debug("socket.timeout witk ikpdb.status=%s", ikpdb.status)
if ikpdb.status == 'terminated':
_logger.n_debug("breaking IKPdbConnectionHandler.receive() "
"network loop as ikpdb state is 'terminated'.")
return {
'command': '_InternalQuit',
'args':{}
}
continue
except socket.error as socket_err:
if ikpdb.status == 'terminated':
return {'command': '_InternalQuit',
'args':{'socket_error_number': socket_err.errno,
'socket_error_str': socket_err.strerror}}
continue
except Exception as exc:
_logger.g_error("Unexecpected Error: '%s' in IKPdbConnectionHandler"
".command_loop.", exc)
_logger.g_error(traceback.format_exc())
print("".join(traceback.format_stack()))
return {
'command': '_InternalQuit',
'args':{
"error": exc.__class__.__name__,
"message": exc.message
}
}
# received data is utf8 encoded
self._received_data += data.decode('utf-8')
# have we received a MAGIC_CODE
try:
magic_code_idx = self._received_data.index(self.MAGIC_CODE)
except ValueError:
continue
# Have we received a 'length='
try:
length_idx = self._received_data.index(u'length=')
except ValueError:
continue
# extract length content from received data
json_length = int(self._received_data[length_idx + 7:magic_code_idx])
message_length = magic_code_idx + len(self.MAGIC_CODE) + json_length
if message_length <= len(self._received_data):
full_message = self._received_data[:message_length]
self._received_data = self._received_data[message_length:]
if len(self._received_data) > 0:
self.SOCKET_BUFFER_SIZE = 0
else:
self.SOCKET_BUFFER_SIZE = 4096
break
else:
self.SOCKET_BUFFER_SIZE = message_length - len(self._received_data)
self.log_received(full_message)
obj = self.decode(full_message)
return obj
##
# Debugger
#
class IKPdbQuit(Exception):
"""A dummy Exception used by debugger to quit debugged program."""
pass
def IKPdbRepr(t):
""" A function that returns a type representation suitable for debugger GUI.
:param t: anyThing
:return: a string representation of t's type
Note: Type representation str format is not finalized...
"""
try:
result = str(type(t)).split('\'')[1]
except:
result = "IKPdbReprError"
return result
class IKBreakpoint(object):
""" IKBreakpoint implements and manages IKP3db Breakpoints.
Basically a breakpoint is described by:
- `number`: a uniq breakpoint number
- `file_name`: using a canonical file path
- `line_number`: 1 based
- `condition`: an optional python expression used to trigger conditional breakpoints.Basically
- `enabled`: a flag to enable / disable the breakpoint
The debugger manages Breakpoints using 3 lists maintained by IKBreakpoint:
- `breakpoints_files` contains all breakpoints line numbers indexed by file_name
- `breakpoints_by_file_and_line` contains all breakpoints indexed by (file, line)
- `breakpoints_by_number` is an indexed list of all breakpoints.
This class also maintains a `any_active_breakpoint` boolean class attribute
that is False when there is no active breakpoint. This flag is used to
trigger `TURBO Mode`.
:param file_name: a CANONICAL file name.
:type file_name: str
:param line_number: breakpoint's line number (1 based).
:type line_number: int
:param condition: an optional python expression used to trigger
conditional breakpoints.
:type condition: str
:param enabled: a flag to enable / disable the breakpoint.
:type enabled: bool
"""
breakpoints_files = {} #: list of lines indexed by canonical file names
breakpoints_by_file_and_line = {} #: list of breakpoints indexed by (file_name, line)
breakpoints_by_number = [] #: list of breakpoints indexed by number.
next_breakpoint_number = 0 #: Used to allocate next breakpoint number.
any_active_breakpoint = False #: False when there is no active breakpoint.
def __init__(self, file_name, line_number, condition=None, enabled=True):
self.file_name = file_name # In canonical form!
self.line_number = line_number
self.condition = condition
self.enabled = enabled
# Allocate number
self.number = IKBreakpoint.next_breakpoint_number
IKBreakpoint.next_breakpoint_number += 1
# update all lists
IKBreakpoint.breakpoints_by_number.append(self)
IKBreakpoint.breakpoints_by_file_and_line[file_name, line_number] = self
IKBreakpoint.breakpoints_files[file_name] = \
IKBreakpoint.breakpoints_files.get(file_name, [])+[line_number]
if enabled:
IKBreakpoint.any_active_breakpoint = True
def clear(self):
""" Clear a breakpoint by removing it from all lists.
"""
del IKBreakpoint.breakpoints_by_file_and_line[self.file_name, self.line_number]
IKBreakpoint.breakpoints_by_number[self.number] = None
IKBreakpoint.breakpoints_files[self.file_name].remove(self.line_number)
if len(IKBreakpoint.breakpoints_files[self.file_name]) == 0:
del IKBreakpoint.breakpoints_files[self.file_name]
IKBreakpoint.update_active_breakpoint_flag()
@classmethod
def update_active_breakpoint_flag(cls):
""" Checks all breakpoints to find wether at least one is active and
update `any_active_breakpoint` accordingly.
"""
cls.any_active_breakpoint=any([bp.enabled for bp in cls.breakpoints_by_number if bp])
@classmethod
def lookup_effective_breakpoint(cls, file_name, line_number, frame):
""" Checks if there is an enabled breakpoint at given file_name and
line_number. Check breakpoint condition if any.
:return: found, enabled and condition verified breakpoint or None
:rtype: IKPdbBreakpoint or None
"""
bp = cls.breakpoints_by_file_and_line.get((file_name, line_number), None)
if not bp:
return None
if not bp.enabled:
return None
if not bp.condition:
return bp
try:
value = eval(bp.condition, frame.f_globals, frame.f_locals)
return bp if value else None
except:
pass
return None
@classmethod
def get_breakpoints_list(cls):
""":return: a list of all breakpoints.
:rtype: a list of dict with this keys: `breakpoint_number`, `bp.number`,
`file_name`, `line_number`, `condition`, `enabled`.
Warning: IKPDb line numbers are 1 based so line number conversion
must be done by clients (eg. inouk.ikpdb for Cloud9)
"""
breakpoints_list = []
for bp in cls.breakpoints_by_number:
if bp: # breakpoint #0 exists and is always None
bp_dict = {
'breakpoint_number': bp.number,
'file_name': bp.file_name,
'line_number': bp.line_number,
'condition': bp.condition,
'enabled': bp.enabled,
}
breakpoints_list.append(bp_dict)
return breakpoints_list
@classmethod
def disable_all_breakpoints(cls):
""" Disable all breakpoints and udate `active_breakpoint_flag`.
"""
for bp in cls.breakpoints_by_number:
if bp: # breakpoint #0 exists and is always None
bp.enabled = False
cls.update_active_breakpoint_flag()
return
@classmethod
def backup_breakpoints_state(cls):
""" Returns the state of all breakpoints in a list that can be used
later to restore all breakpoints state"""
all_breakpoints_state = []
for bp in cls.breakpoints_by_number:
if bp:
all_breakpoints_state.append((bp.number,
bp.enabled,
bp.condition,))
return all_breakpoints_state
@classmethod
def restore_breakpoints_state(cls, breakpoints_state_list):
"""Restore the state of breakpoints given a list provided by
backup_breakpoints_state(). If list of breakpoint has changed
since backup missing or added breakpoints are ignored.
breakpoints_state_list is a list of tuple. Each tuple is of form:
(breakpoint_number, enabled, condition)
"""
for breakpoint_state in breakpoints_state_list:
bp = cls.breakpoints_by_number[breakpoint_state[0]]
if bp:
bp.enabled = breakpoint_state[1]
bp.condition = breakpoint_state[2]
cls.update_active_breakpoint_flag()
return
class IKPdb(object):
""" Main debugger class.
:param skip: reserved for future use
:param working_directory: allows to force debugger's Current Working
Directory (CWD). `working_directory` is used for file
mapping between IKPdb and clients.
`working_directory` is concatenated with file path
exchanged with debugger's client to get absolute
file's paths.
:type working_directory: str
:param stop_at_first_statement: defines wether debugger must break at
first statement. None don't break, else break.
:type stop_at_first_statement: str
Take note that, right now, IKPdb is used as singleton.
"""
def __init__(self, skip=None, stop_at_first_statement=False, working_directory=None,
client_working_directory=None):
# TODO: manage skip
self.skip = set(skip) if skip else None
self.debugger_thread_ident = None
self.file_name_cache = {}
self._CWD = working_directory or os.getcwd()
if self._CWD and self._CWD[-1] != '/':
self._CWD += '/'
self._CLIENT_CWD = client_working_directory or ''
if self._CLIENT_CWD and self._CLIENT_CWD[-1] != '/':
self._CLIENT_CWD += '/'
self.mainpyfile = ''
self._active_breakpoint_lock = threading.Lock()
self._active_thread_lock = threading.Lock()
self._command_q = queue.Queue(maxsize=1)
# tracing is disabled until required
self.execution_started = False
self.tracing_enabled = False
# At any time IKP3db status can be
# * 'pending' => Execution has not started yet
# * 'running'
# * 'stopped' => either on a breakpoint or an exception
# * 'terminated'
self.status = 'pending'
self.debugged_thread_ident = None
self.debugged_thread_name = None
# stop management
self.pending_stop = False # True if any of frame_xxxx is set
self.frame_stop = None # stepOver and stepInto
self.frame_calling = None # stepInto
self.frame_return = None # stepOut and stepOver
self.frame_suspend = False # If true, debugger will stop at next frame
# last frame to dump ; allows to dump only debugged program frames
self.frame_beginning = None
# If True, debugger breaks on first line to allow user to setup
# some breakpoints.
self.stop_at_first_statement = True if stop_at_first_statement else False
# Some parameters that may need to become cli options
self.CGI_ESCAPE_EVALUATE_OUTPUT = False
def canonic(self, file_name):
""" returns canonical version of a file name.
A canonical file name is an absolute, lowercase normalized path
to a given file.
"""
if file_name == "<" + file_name[1:-1] + ">":
return file_name
c_file_name = self.file_name_cache.get(file_name)
if not c_file_name:
c_file_name = os.path.abspath(file_name)
c_file_name = os.path.normcase(c_file_name)
self.file_name_cache[file_name] = c_file_name
return c_file_name
def normalize_path_in(self, client_file_name):
"""Translate a (possibly incomplete) file or module name received from debugging client
into an absolute file name.
"""
_logger.p_debug("normalize_path_in(%s) with os.getcwd()=>%s", client_file_name, os.getcwd())
# remove client CWD from file_path
if client_file_name.startswith(self._CLIENT_CWD):
file_name = client_file_name[len(self._CLIENT_CWD):]
else:
file_name = client_file_name
# Try to find file using it's absolute path
if os.path.isabs(file_name) and os.path.exists(file_name):
_logger.p_debug(" => found absolute path: '%s'", file_name)
return file_name
# Can we find the file relatively to launch CWD (useful with buildout)
f = os.path.join(self._CWD, file_name)
if os.path.exists(f):
_logger.p_debug(" => found path relative to self._CWD: '%s'", f)
return f
# Can we find file relatively to launch script
f = os.path.join(sys.path[0], file_name)
if os.path.exists(f) and self.canonic(f) == self.mainpyfile:
_logger.p_debug(" => found path relative to launch script: '%s'", f)
return f
# Try as an absolute path after adding .py extension
root, ext = os.path.splitext(file_name)
if ext == '':
f = file_name + '.py'
if os.path.isabs(f):
_logger.p_debug(" => found absolute path after adding .py extension: '%s'", f)
return f
# Can we find the file in system path
for dir_name in sys.path:
while os.path.islink(dir_name):
dir_name = os.readlink(dir_name)
f = os.path.join(dir_name, file_name)
if os.path.exists(f):
_logger.p_debug(" => found path in sys.path: '%s'", f)
return f
return None
def normalize_path_out(self, path):
"""Normalizes path sent to client
:param path: path to normalize
:return: normalized path
"""
if path.startswith(self._CWD):
normalized_path = path[len(self._CWD):]
else:
normalized_path = path
# For remote debugging preprend client CWD
if self._CLIENT_CWD:
normalized_path = os.path.join(self._CLIENT_CWD, normalized_path)
_logger.p_debug("normalize_path_out('%s') => %s", path, normalized_path)
return normalized_path
def object_properties_count(self, o):
""" returns the number of user browsable properties of an object. """
o_type = type(o)
if isinstance(o, (dict, list, tuple, set)):
return len(o)
elif isinstance(o, (type(None), bool, float,
str, int,
bytes, types.ModuleType,
types.MethodType, types.FunctionType)):
return 0
else:
# Following lines are used to debug variables members browsing
# and counting
# if False and str(o_type) == "<class 'socket._socketobject'>":
# print "@378"
# print dir(o)
# print "hasattr(o, '__dict__')=%s" % hasattr(o,'__dict__')
# count = 0
# if hasattr(o, '__dict__'):
# for m_name, m_value in o.__dict__.iteritems():
# if m_name.startswith('__'):
# print " %s=>False" % (m_name,)
# continue
# if type(m_value) in (types.ModuleType, types.MethodType, types.FunctionType,):
# print " %s=>False" % (m_name,)
# continue
# print " %s=>True" % (m_name,)
# count +=1
# print " %s => %s = %s" % (o, count, dir(o),)
# else:
try:
if hasattr(o, '__dict__'):
count = len([m_name for m_name, m_value in o.__dict__.items()
if not m_name.startswith('__')
and not type(m_value) in (types.ModuleType,
types.MethodType,
types.FunctionType,) ])
else:
count = 0
except:
# Thank you werkzeug __getattr__ overloading!
count = 0
return count
def extract_object_properties(self, o, limit_size=False):
"""Extracts all properties from an object (eg. f_locals, f_globals,
user dict, instance ...) and returns them as an array of variables.
"""
try:
prop_str = repr(o)[:512]
except:
prop_str = "Error while extracting value"
_logger.e_debug("extract_object_properties(%s)", prop_str)
var_list = []
if isinstance(o, dict):
a_var_name = None
a_var_value = None
for a_var_name in o:
a_var_value = o[a_var_name]
children_count = self.object_properties_count(a_var_value)
v_name, v_value, v_type = self.extract_name_value_type(a_var_name,
a_var_value,
limit_size=limit_size)
a_var_info = {
'id': id(a_var_value),
'name': v_name,
'type': "%s%s" % (v_type, " [%s]" % children_count if children_count else '',),
'value': v_value,
'children_count': children_count,
}
var_list.append(a_var_info)
elif type(o) in (list, tuple, set,):
MAX_CHILDREN_TO_RETURN = 256
MAX_CHILDREN_MESSAGE = "Truncated by ikpdb (don't hot change me !)."
a_var_name = None
a_var_value = None
do_truncate = len(o) > MAX_CHILDREN_TO_RETURN
for idx, a_var_value in enumerate(o):
children_count = self.object_properties_count(a_var_value)
v_name, v_value, v_type = self.extract_name_value_type(idx,
a_var_value,
limit_size=limit_size)
var_list.append({
'id': id(a_var_value),
'name': v_name,
'type': "%s%s" % (v_type, " [%s]" % children_count if children_count else '',),
'value': v_value,
'children_count': children_count,
})
if do_truncate and idx==MAX_CHILDREN_TO_RETURN-1:
var_list.append({
'id': None,
'name': str(MAX_CHILDREN_TO_RETURN),
'type': '',
'value': MAX_CHILDREN_MESSAGE,
'children_count': 0,
})
break
else:
a_var_name = None
a_var_value = None
if hasattr(o, '__dict__'):
for a_var_name, a_var_value in o.__dict__.items():
if (not a_var_name.startswith('__')
and not type(a_var_value) in (types.ModuleType,
types.MethodType,
types.FunctionType,)):
children_count = self.object_properties_count(a_var_value)
v_name, v_value, v_type = self.extract_name_value_type(a_var_name,
a_var_value,
limit_size=limit_size)
var_list.append({
'id': id(a_var_value),
'name': v_name,
'type': "%s%s" % (v_type, " [%s]" % children_count if children_count else '',),
'value': v_value,
'children_count': children_count,
})
return var_list
def extract_name_value_type(self, name, value, limit_size=False):
"""Extracts value of any object, eventually reduces it's size and
returns name, truncated value and type (for str with size appended)
"""
MAX_STRING_LEN_TO_RETURN = 487
try:
t_value = repr(value)
except:
t_value = "Error while extracting value"
# convert all var names to string
if isinstance(name, str):
r_name = name
else:
r_name = repr(name)
# truncate value to limit data flow between ikpdb and client
if len(t_value) > MAX_STRING_LEN_TO_RETURN:
r_value = "%s ... (truncated by ikpdb)" % (t_value[:MAX_STRING_LEN_TO_RETURN],)
r_name = "%s*" % r_name # add a visual marker to truncated var's name
else:
r_value = t_value
if isinstance(value, str):
r_type = "%s [%s]" % (IKPdbRepr(value), len(value),)
else:
r_type = IKPdbRepr(value)
return r_name, r_value, r_type
def dump_frames(self, frame):
""" dumps frames chain in a representation suitable for serialization
and remote (debugger) client usage.
"""
current_thread = threading.currentThread()
frames = []
frame_browser = frame
# Browse the frame chain as far as we can
_logger.f_debug("dump_frames(), frame analysis:")
spacer = ""
while hasattr(frame_browser, 'f_back') and frame_browser.f_back != self.frame_beginning:
spacer += "="
_logger.f_debug("%s>frame = %s, frame.f_code = %s, frame.f_back = %s, "
"self.frame_beginning = %s",
spacer,
hex(id(frame_browser)),
frame_browser.f_code,
hex(id(frame_browser.f_back)),
hex(id(self.frame_beginning)))
# At root frame, globals == locals so we dump only globals
if hasattr(frame_browser.f_back, 'f_back')\
and frame_browser.f_back.f_back != self.frame_beginning:
locals_vars_list = self.extract_object_properties(frame_browser.f_locals,
limit_size=True)
else:
locals_vars_list = []
globals_vars_list = self.extract_object_properties(frame_browser.f_globals,
limit_size=True)
# normalize path sent to debugging client
file_path = self.normalize_path_out(frame_browser.f_code.co_filename)
frame_name = "%s() [%s]" % (frame_browser.f_code.co_name, current_thread.name,)
remote_frame = {
'id': id(frame_browser),
'name': frame_name,
'line_number': frame_browser.f_lineno, # Warning 1 based
'file_path': file_path,
'f_locals': locals_vars_list + globals_vars_list,
'thread': current_thread.ident,
'thread_name': current_thread.name
}
frames.append(remote_frame)
frame_browser = frame_browser.f_back
return frames
def evaluate(self, frame_id, expression, global_context=False, disable_break=False):
"""Evaluates 'expression' in the context of the frame identified by
'frame_id' or globally.
Breakpoints are disabled depending on 'disable_break' value.
Returns a tuple of value and type both as str.
Note that - depending on the CGI_ESCAPE_EVALUATE_OUTPUT attribute - value is
escaped.
"""
if disable_break:
breakpoints_backup = IKBreakpoint.backup_breakpoints_state()
IKBreakpoint.disable_all_breakpoints()
if frame_id and not global_context:
eval_frame = ctypes.cast(frame_id, ctypes.py_object).value
global_vars = eval_frame.f_globals
local_vars = eval_frame.f_locals
else:
global_vars = None
local_vars = None
try:
result = eval(expression, global_vars, local_vars)
result_type = IKPdbRepr(result)
result_value = repr(result)
except SyntaxError:
# eval() failed, try with exec to handle statements
try:
result = exec(expression, global_vars, local_vars)
result_type = IKPdbRepr(result)
result_value = repr(result)
except Exception as e:
t, result = sys.exc_info()[:2]
if isinstance(t, str):
result_type = t
else:
result_type = str(t.__name__)
result_value = "%s: %s" % (result_type, result,)
except:
t, result = sys.exc_info()[:2]
if isinstance(t, str):
result_type = t
else:
result_type = t.__name__
result_value = "%s: %s" % (result_type, result,)
if disable_break:
IKBreakpoint.restore_breakpoints_state(breakpoints_backup)
_logger.e_debug("evaluate(%s) => result_value=%s, result_type=%s, result=%s",
expression,
result_value,
result_type,
result)
if self.CGI_ESCAPE_EVALUATE_OUTPUT:
result_value = cgi.escape(result_value)
# We must check that result is json.dump compatible so that it can be sent back to client.
try:
json.dumps(result_value)
except:
t, result = sys.exc_info()[:2]
if isinstance(t, str):
result_type = t
else:
result_type = t.__name__
result_value = "<plaintext>%s: IKP3db is unable to JSON encode result to send it to "\
"debugging client.\n"\
" This typically occurs if you try to print a string that cannot be"\
" decoded to 'UTF-8'.\n"\
" You should be able to evaluate result and inspect it's content"\
" by removing the print statement." % result_type
return result_value, result_type
def let_variable(self, frame_id, var_name, expression_value):
""" Let a frame's var with a value by building then eval a let
expression with breakoints disabled.
"""
breakpoints_backup = IKBreakpoint.backup_breakpoints_state()
IKBreakpoint.disable_all_breakpoints()