mirror of
https://github.com/Gallopsled/pwntools.git
synced 2026-10-12 02:02:27 +00:00
Scott Laglerandpeace-maker
5efe909a91
* Add Type Annotations for parts of pwnlib/util * fix: use Callable in pwnlib/util/lists.py * fix: Import Errors * fix: use getLogger in pwnlib/util/net * feat: use collections.abc.Buffer where appropriate * fix: remove collections.abc.Buffer to use Sequence * add CHANGELOG.md entry * Improve packing and fiddling types * Introduce own `BytesLike` type instead of 3.12 Buffer * Improve web.wget types * Use TypedDict in util.net Specify generics properly. * Improve pwnlib.proc types * Improve pwnlib.util.lists types * Fix util.splash types * Make TypeVars private * Improve util.hexdump types * Remove partial types from packing.fit/flat Those require more work to be useful. * Fix implicit wget return * Fix term.text type export * Fix js_escape return type * Fix sh_string doctest types * Fix usage of ssh_channel as ssh_process We access the pid, so a plain ssh_channel is not enough. * Improve util.lists types * Improve packing.dd types * Fix relying on socket.AddressFamily type at runtime The type is only in typeshed and not the real socket module. Don't access it at runtime, only at type checking. * Fix lists.findall with string needle * Fix dd on two files * Update mypy-baseline.txt * Remove testing artifacts from mypy-baseline.txt --------- Co-authored-by: peace-maker <[email protected]>
717 lines
26 KiB
Python
717 lines
26 KiB
Python
"""
|
|
Logging module for printing status during an exploit, and internally
|
|
within ``pwntools``.
|
|
|
|
Exploit Developers
|
|
------------------
|
|
By using the standard ``from pwn import *``, an object named ``log`` will
|
|
be inserted into the global namespace. You can use this to print out
|
|
status messages during exploitation.
|
|
|
|
For example,::
|
|
|
|
log.info('Hello, world!')
|
|
|
|
prints::
|
|
|
|
[*] Hello, world!
|
|
|
|
Additionally, there are some nifty mechanisms for performing status updates
|
|
on a running job (e.g. when brute-forcing).::
|
|
|
|
p = log.progress('Working')
|
|
p.status('Reticulating splines')
|
|
time.sleep(1)
|
|
p.success('Got a shell!')
|
|
|
|
|
|
The verbosity of logging can be most easily controlled by setting
|
|
``log_level`` on the global ``context`` object.::
|
|
|
|
log.info("No you see me")
|
|
context.log_level = 'error'
|
|
log.info("Now you don't")
|
|
|
|
The purpose of this attribute is to control what gets printed to the screen,
|
|
not what gets emitted. This means that you can put all logging events into
|
|
a log file, while only wanting to see a small subset of them on your screen.
|
|
|
|
Pwnlib Developers
|
|
-----------------
|
|
A module-specific logger can be imported into the module via::
|
|
|
|
from pwnlib.log import getLogger
|
|
log = getLogger(__name__)
|
|
|
|
This provides an easy way to filter logging programmatically
|
|
or via a configuration file for debugging.
|
|
|
|
When using ``progress``, you should use the ``with``
|
|
keyword to manage scoping, to ensure the spinner stops if an
|
|
exception is thrown.
|
|
|
|
Technical details
|
|
-----------------
|
|
Familiarity with the :mod:`logging` module is assumed.
|
|
|
|
A pwnlib root logger named 'pwnlib' is created and a custom handler and
|
|
formatter is installed for it. The handler determines its logging level from
|
|
:data:`context.log_level`.
|
|
|
|
Ideally :data:`context.log_level` should only affect which records will be
|
|
emitted by the handler such that e.g. logging to a file will not be changed by
|
|
it. But for performance reasons it is not feasible log everything in the normal
|
|
case. In particular there are tight loops inside :mod:`pwnlib.tubes.tube`, which
|
|
we would like to be able to debug, but if we are not debugging them, they should
|
|
not spit out messages (even to a log file). For this reason there are a few places
|
|
inside pwnlib, that will not even emit a record without :data:`context.log_level`
|
|
being set to `logging.DEBUG` or below.
|
|
|
|
Log records created by ``Progress`` and ``Logger`` objects will set
|
|
``'pwnlib_msgtype'`` on the ``extra`` field to signal which kind of message was
|
|
generated. This information is used by the formatter to prepend a symbol to the
|
|
message, e.g. ``'[+] '`` in ``'[+] got a shell!'``
|
|
|
|
This field is ignored when using the ``logging`` module's standard formatters.
|
|
|
|
All status updates (which are not dropped due to throttling) on progress loggers
|
|
result in a log record being created. The ``extra`` field then carries a
|
|
reference to the ``Progress`` logger as ``'pwnlib_progress'``.
|
|
|
|
If the custom handler determines that :data:`term.term_mode` is enabled, log
|
|
records that have a ``'pwnlib_progess'`` in their ``extra`` field will not
|
|
result in a message being emitted but rather an animated progress line (with a
|
|
spinner!) being created. Note that other handlers will still see a meaningful
|
|
log record.
|
|
|
|
The custom handler will only handle log records with a level of at least
|
|
:data:`context.log_level`. Thus if e.g. the level for the
|
|
``'pwnlib.tubes.ssh'`` is set to ``'DEBUG'`` no additional output will show up
|
|
unless :data:`context.log_level` is also set to ``'DEBUG'``. Other handlers
|
|
will however see the extra log records generated by the ``'pwnlib.tubes.ssh'``
|
|
logger.
|
|
"""
|
|
from __future__ import annotations
|
|
from io import TextIOWrapper
|
|
import logging
|
|
import os
|
|
import random
|
|
import string
|
|
import threading
|
|
import time
|
|
from types import TracebackType
|
|
from typing import Any, NoReturn
|
|
|
|
from pwnlib import term
|
|
from pwnlib.config import register_config
|
|
from pwnlib.context import Thread
|
|
from pwnlib.context import context
|
|
from pwnlib.exception import PwnlibException
|
|
from pwnlib.term import spinners
|
|
from pwnlib.term import text
|
|
|
|
__all__ = [
|
|
'getLogger', 'install_default_handler', 'rootlogger'
|
|
]
|
|
|
|
|
|
|
|
# list of prefixes to use for the different message types. note that the `text`
|
|
# module won't add any escape codes if `pwnlib.context.log_console.isatty()` is `False`
|
|
_msgtype_prefixes = {
|
|
'status' : [text.magenta, 'x'],
|
|
'success' : [text.bold_green, '+'],
|
|
'failure' : [text.bold_red, '-'],
|
|
'debug' : [text.bold_red, 'DEBUG'],
|
|
'info' : [text.bold_blue, '*'],
|
|
'warning' : [text.bold_yellow, '!'],
|
|
'error' : [text.on_red, 'ERROR'],
|
|
'exception' : [text.on_red, 'ERROR'],
|
|
'critical' : [text.on_red, 'CRITICAL'],
|
|
'info_once' : [text.bold_blue, '*'],
|
|
'warning_once' : [text.bold_yellow, '!'],
|
|
}
|
|
|
|
|
|
def read_log_config(settings: dict[str, Any]) -> None:
|
|
log = getLogger(__name__)
|
|
for key, value in settings.items():
|
|
if '.' not in key:
|
|
log.warn("Invalid configuration option %r in section %r" % (key, 'log'))
|
|
continue
|
|
value = str(value)
|
|
|
|
msgtype, key = key.split('.', 1)
|
|
|
|
if key == 'color':
|
|
current = _msgtype_prefixes[msgtype][0]
|
|
_msgtype_prefixes[msgtype][0] = getattr(text, value, current)
|
|
|
|
elif key == 'symbol':
|
|
_msgtype_prefixes[msgtype][1] = value
|
|
|
|
else:
|
|
log.warn("Unknown configuration option %r in section %r" % (key, 'log'))
|
|
|
|
register_config('log', read_log_config)
|
|
|
|
# the text decoration to use for spinners. the spinners themselves can be found
|
|
# in the `pwnlib.term.spinners` module
|
|
_spinner_style = text.bold_blue
|
|
|
|
class Progress:
|
|
"""
|
|
Progress logger used to generate log records associated with some running
|
|
job. Instances can be used as context managers which will automatically
|
|
declare the running job a success upon exit or a failure upon a thrown
|
|
exception. After :meth:`success` or :meth:`failure` is called the status
|
|
can no longer be updated.
|
|
|
|
This class is intended for internal use. Progress loggers should be created
|
|
using :meth:`Logger.progress`.
|
|
"""
|
|
def __init__(self, logger: Logger, msg: str, status: str, level: int, args: tuple[Any], kwargs: dict[str, Any]) -> None:
|
|
self._logger = logger
|
|
self._msg = msg
|
|
self._status = status
|
|
self._level = level
|
|
self._stopped = False
|
|
self.last_status = 0.0
|
|
self.rate = kwargs.pop('rate', 0)
|
|
self._log(status, args, kwargs, 'status')
|
|
# it is a common use case to create a logger and then immediately update
|
|
# its status line, so we reset `last_status` to accommodate this pattern
|
|
self.last_status = 0.0
|
|
|
|
def _log(self, status: object, args: tuple[Any], kwargs: dict[str, Any], msgtype: str) -> None:
|
|
# Logs are strings, not bytes. Handle Python3 bytes() objects.
|
|
status = _need_text(status)
|
|
|
|
# this progress logger is stopped, so don't generate any more records
|
|
if self._stopped:
|
|
return
|
|
msg = self._msg
|
|
if msg and status:
|
|
msg += ': '
|
|
msg += status
|
|
self._logger._log(self._level, msg, args, kwargs, msgtype, self)
|
|
|
|
def status(self, status: str, *args: Any, **kwargs: Any) -> None:
|
|
"""status(status, *args, **kwargs)
|
|
|
|
Logs a status update for the running job.
|
|
|
|
If the progress logger is animated the status line will be updated in
|
|
place.
|
|
|
|
Status updates can be throttled at one update per `rate` seconds.
|
|
"""
|
|
now = time.time()
|
|
if (now - self.last_status) > self.rate:
|
|
self.last_status = now
|
|
self._log(status, args, kwargs, 'status')
|
|
|
|
def success(self, status: object = 'Done', *args: Any, **kwargs: Any) -> None:
|
|
"""success(status = 'Done', *args, **kwargs)
|
|
|
|
Logs that the running job succeeded. No further status updates are
|
|
allowed.
|
|
|
|
If the Logger is animated, the animation is stopped.
|
|
"""
|
|
self._log(status, args, kwargs, 'success')
|
|
self._stopped = True
|
|
|
|
def failure(self, status: object = 'Failed', *args: Any, **kwargs: Any) -> None:
|
|
"""failure(message)
|
|
|
|
Logs that the running job failed. No further status updates are
|
|
allowed.
|
|
|
|
If the Logger is animated, the animation is stopped.
|
|
"""
|
|
self._log(status, args, kwargs, 'failure')
|
|
self._stopped = True
|
|
|
|
def __enter__(self) -> Progress:
|
|
return self
|
|
|
|
def __exit__(self, exc_typ: type[BaseException], exc_val: BaseException, exc_tb: TracebackType | None) -> None:
|
|
# if the progress logger is already stopped these are no-ops
|
|
if exc_typ is None:
|
|
self.success()
|
|
else:
|
|
self.failure()
|
|
|
|
class Logger:
|
|
"""
|
|
A class akin to the :class:`logging.LoggerAdapter` class. All public
|
|
methods defined on :class:`logging.Logger` instances are defined on this
|
|
class.
|
|
|
|
Also adds some ``pwnlib`` flavor:
|
|
|
|
* :meth:`progress` (alias :meth:`waitfor`)
|
|
* :meth:`success`
|
|
* :meth:`failure`
|
|
* :meth:`indented`
|
|
* :meth:`info_once`
|
|
* :meth:`warning_once` (alias :meth:`warn_once`)
|
|
|
|
Adds ``pwnlib``-specific information for coloring, indentation and progress
|
|
logging via log records ``extra`` field.
|
|
|
|
Loggers instantiated with :func:`getLogger` will be of this class.
|
|
"""
|
|
_one_time_infos: set[str] = set()
|
|
_one_time_warnings: set[str] = set()
|
|
|
|
def __init__(self, logger: logging.Logger | None = None):
|
|
if logger is None:
|
|
# This is a minor hack to permit user-defined classes which inherit
|
|
# from a tube (which do not actually reside in the pwnlib library)
|
|
# to receive logging abilities that behave as they would expect from
|
|
# the rest of the library
|
|
module = self.__module__
|
|
if not module.startswith('pwnlib'):
|
|
module = 'pwnlib.' + module
|
|
# - end hack -
|
|
|
|
logger_name = '%s.%s.%s' % (module, self.__class__.__name__, id(self))
|
|
logger = logging.getLogger(logger_name)
|
|
|
|
self._logger = logger
|
|
|
|
def _getlevel(self, levelString: str | int) -> int:
|
|
if isinstance(levelString, int):
|
|
return levelString
|
|
return logging._nameToLevel[levelString.upper()]
|
|
|
|
|
|
def _log(self, level: int, msg: object, args: tuple[Any], kwargs: dict[str, Any], msgtype: str | None, progress: Progress | None = None):
|
|
# Logs are strings, not bytes. Handle Python3 bytes() objects.
|
|
msg = _need_text(msg)
|
|
|
|
extra = kwargs.get('extra', {})
|
|
extra.setdefault('pwnlib_msgtype', msgtype)
|
|
extra.setdefault('pwnlib_progress', progress)
|
|
kwargs['extra'] = extra
|
|
self._logger.log(level, msg, *args, **kwargs)
|
|
|
|
def progress(self, message: str, status: str = '', *args: Any, **kwargs: Any) -> Progress:
|
|
"""progress(message, status = '', *args, level = logging.INFO, **kwargs) -> Progress
|
|
|
|
Creates a new progress logger which creates log records with log level
|
|
``level``.
|
|
|
|
An optional ``rate`` parameter can be specified to control the frequency of status updates.
|
|
Calls to :meth:`Progress.status` will be throttled to at most one update per `rate` seconds.
|
|
Throtteled status updates will not be emitted at all, so they will not show up in log files.
|
|
Use this to avoid generating too many log records in tight loops.
|
|
|
|
Progress status can be updated using :meth:`Progress.status` and stopped
|
|
using :meth:`Progress.success` or :meth:`Progress.failure`.
|
|
|
|
If `term.term_mode` is enabled the progress logger will be animated.
|
|
|
|
The progress manager also functions as a context manager. Using context
|
|
managers ensures that animations stop even if an exception is raised.
|
|
|
|
.. code-block:: python
|
|
|
|
with log.progress('Trying something...') as p:
|
|
for i in range(10):
|
|
p.status("At %i" % i)
|
|
time.sleep(0.5)
|
|
x = 1/0
|
|
"""
|
|
level = self._getlevel(kwargs.pop('level', logging.INFO))
|
|
return Progress(self, message, status, level, args, kwargs)
|
|
|
|
def waitfor(self, *args: Any, **kwargs: Any) -> Progress:
|
|
"""Alias for :meth:`progress`."""
|
|
return self.progress(*args, **kwargs)
|
|
|
|
def indented(self, message: object, *args: Any, **kwargs: Any) -> None:
|
|
"""indented(message, *args, level = logging.INFO, **kwargs)
|
|
|
|
Log a message but don't put a line prefix on it.
|
|
|
|
Arguments:
|
|
level(int): Alternate log level at which to set the indented
|
|
message. Defaults to :const:`logging.INFO`.
|
|
"""
|
|
level = self._getlevel(kwargs.pop('level', logging.INFO))
|
|
self._log(level, message, args, kwargs, 'indented')
|
|
|
|
def success(self, message: object, *args: Any, **kwargs: Any) -> None:
|
|
"""success(message, *args, **kwargs)
|
|
|
|
Logs a success message.
|
|
"""
|
|
self._log(logging.INFO, message, args, kwargs, 'success')
|
|
|
|
def failure(self, message: object, *args: Any, **kwargs: Any) -> None:
|
|
"""failure(message, *args, **kwargs)
|
|
|
|
Logs a failure message.
|
|
"""
|
|
self._log(logging.INFO, message, args, kwargs, 'failure')
|
|
|
|
def info_once(self, message: object, *args: Any, **kwargs: Any) -> None:
|
|
"""info_once(message, *args, **kwargs)
|
|
|
|
Logs an info message. The same message is never printed again.
|
|
"""
|
|
m = _need_text(message) % args
|
|
if m not in self._one_time_infos:
|
|
if self.isEnabledFor(logging.INFO):
|
|
self._one_time_infos.add(m)
|
|
self._log(logging.INFO, message, args, kwargs, 'info_once')
|
|
|
|
def warning_once(self, message: object, *args: Any, **kwargs: Any) -> None:
|
|
"""warning_once(message, *args, **kwargs)
|
|
|
|
Logs a warning message. The same message is never printed again.
|
|
"""
|
|
m = _need_text(message) % args
|
|
if m not in self._one_time_warnings:
|
|
if self.isEnabledFor(logging.WARNING):
|
|
self._one_time_warnings.add(m)
|
|
self._log(logging.WARNING, message, args, kwargs, 'warning_once')
|
|
|
|
def warn_once(self, *args: Any, **kwargs: Any) -> None:
|
|
"""Alias for :meth:`warning_once`."""
|
|
return self.warning_once(*args, **kwargs)
|
|
|
|
# logging functions also exposed by `logging.Logger`
|
|
|
|
def debug(self, message: object, *args: Any, **kwargs: Any) -> None:
|
|
"""debug(message, *args, **kwargs)
|
|
|
|
Logs a debug message.
|
|
"""
|
|
self._log(logging.DEBUG, message, args, kwargs, 'debug')
|
|
|
|
def info(self, message: object, *args: Any, **kwargs: Any) -> None:
|
|
"""info(message, *args, **kwargs)
|
|
|
|
Logs an info message.
|
|
"""
|
|
self._log(logging.INFO, message, args, kwargs, 'info')
|
|
|
|
def hexdump(self, message: bytes, *args: Any, **kwargs: Any) -> None:
|
|
# cyclic dependencies FTW!
|
|
# TODO: Move pwnlib.util.fiddling.hexdump into a new module.
|
|
import pwnlib.util.fiddling
|
|
|
|
self.info(pwnlib.util.fiddling.hexdump(message, *args, **kwargs))
|
|
|
|
def maybe_hexdump(self, message: bytes, *args: Any, **kwargs: Any) -> None:
|
|
"""maybe_hexdump(self, message, *args, **kwargs)
|
|
|
|
Logs a message using indented. Repeated single byte is compressed, and
|
|
unprintable message is hexdumped.
|
|
"""
|
|
if len(set(message)) == 1 and len(message) > 1:
|
|
self.indented('%r * %#x' % (message[:1], len(message)), *args, **kwargs)
|
|
elif len(message) == 1 or all(c in string.printable.encode() for c in message):
|
|
for line in message.splitlines(True):
|
|
self.indented(repr(line), *args, **kwargs)
|
|
else:
|
|
import pwnlib.util.fiddling
|
|
self.indented(pwnlib.util.fiddling.hexdump(message), *args, **kwargs)
|
|
|
|
def warning(self, message: object, *args: Any, **kwargs: Any) -> None:
|
|
"""warning(message, *args, **kwargs)
|
|
|
|
Logs a warning message.
|
|
"""
|
|
self._log(logging.WARNING, message, args, kwargs, 'warning')
|
|
|
|
def warn(self, *args: Any, **kwargs: Any) -> None:
|
|
"""Alias for :meth:`warning`."""
|
|
return self.warning(*args, **kwargs)
|
|
|
|
def error(self, message: object, *args: Any, **kwargs: Any) -> NoReturn:
|
|
"""error(message, *args, **kwargs)
|
|
|
|
To be called outside an exception handler.
|
|
|
|
Logs an error message, then raises a ``PwnlibException``.
|
|
"""
|
|
msg = _need_text(message)
|
|
self._log(logging.ERROR, msg, args, kwargs, 'error')
|
|
raise PwnlibException(msg % args)
|
|
|
|
def exception(self, message: object, *args: Any, **kwargs: Any) -> NoReturn:
|
|
"""exception(message, *args, **kwargs)
|
|
|
|
To be called from an exception handler.
|
|
|
|
Logs a error message, then re-raises the current exception.
|
|
"""
|
|
kwargs["exc_info"] = 1
|
|
self._log(logging.ERROR, message, args, kwargs, 'exception')
|
|
raise
|
|
|
|
def critical(self, message: object, *args: Any, **kwargs: Any) -> None:
|
|
"""critical(message, *args, **kwargs)
|
|
|
|
Logs a critical message.
|
|
"""
|
|
self._log(logging.CRITICAL, message, args, kwargs, 'critical')
|
|
|
|
def log(self, level: int, message: object, *args: Any, **kwargs: Any) -> None:
|
|
"""log(level, message, *args, **kwargs)
|
|
|
|
Logs a message with log level `level`. The ``pwnlib`` formatter will
|
|
use the default :mod:`logging` formater to format this message.
|
|
"""
|
|
self._log(level, message, args, kwargs, None)
|
|
|
|
def isEnabledFor(self, level: int) -> bool:
|
|
"""isEnabledFor(level) -> bool
|
|
|
|
See if the underlying logger is enabled for the specified level.
|
|
"""
|
|
effectiveLevel = self._logger.getEffectiveLevel()
|
|
|
|
if effectiveLevel == 1:
|
|
effectiveLevel = context.log_level
|
|
return effectiveLevel <= level
|
|
|
|
def setLevel(self, level: int | str) -> None:
|
|
"""setLevel(level)
|
|
|
|
Set the logging level for the underlying logger.
|
|
"""
|
|
with context.local(log_level=level):
|
|
self._logger.setLevel(context.log_level)
|
|
|
|
def addHandler(self, handler: Handler) -> None:
|
|
"""addHandler(handler)
|
|
|
|
Add the specified handler to the underlying logger.
|
|
"""
|
|
self._logger.addHandler(handler)
|
|
|
|
def removeHandler(self, handler: Handler) -> None:
|
|
"""removeHandler(handler)
|
|
|
|
Remove the specified handler from the underlying logger.
|
|
"""
|
|
self._logger.removeHandler(handler)
|
|
|
|
@property
|
|
def level(self) -> int:
|
|
return self._logger.level
|
|
@level.setter
|
|
def level(self, value: int | str) -> None:
|
|
with context.local(log_level=value):
|
|
self._logger.level = context.log_level
|
|
|
|
|
|
class Handler(logging.StreamHandler):
|
|
"""
|
|
A custom handler class. This class will report whatever
|
|
:data:`context.log_level` is currently set to as its log level.
|
|
|
|
If :data:`term.term_mode` is enabled log records originating from a progress
|
|
logger will not be emitted but rather an animated progress line will be
|
|
created.
|
|
|
|
An instance of this handler is added to the ``'pwnlib'`` logger.
|
|
"""
|
|
@property
|
|
def stream(self):
|
|
return context.log_console
|
|
@stream.setter
|
|
def stream(self, value):
|
|
pass
|
|
def emit(self, record):
|
|
"""
|
|
Emit a log record or create/update an animated progress logger
|
|
depending on whether :data:`term.term_mode` is enabled.
|
|
"""
|
|
# We have set the root 'pwnlib' logger to have a logLevel of 1,
|
|
# when logging has been enabled via install_default_handler.
|
|
#
|
|
# If the level is 1, we should only process the record if
|
|
# context.log_level is less than the record's log level.
|
|
#
|
|
# If the level is not 1, somebody else expressly set the log
|
|
# level somewhere on the tree, and we should use that value.
|
|
level = logging.getLogger(record.name).getEffectiveLevel()
|
|
if level == 1:
|
|
level = context.log_level
|
|
if level > record.levelno:
|
|
return
|
|
|
|
progress = getattr(record, 'pwnlib_progress', None)
|
|
|
|
# if the record originates from a `Progress` object and term handling
|
|
# is enabled we can have animated spinners! so check that
|
|
# Don't show spinners if the output is being redirected
|
|
if progress is None or not term.term_mode or not os.path.sameopenfile(context.log_console.fileno(), term.term.fd.fileno()):
|
|
super(Handler, self).emit(record)
|
|
return
|
|
|
|
# yay, spinners!
|
|
|
|
# since we want to be able to update the spinner we overwrite the
|
|
# message type so that the formatter doesn't output a prefix symbol
|
|
msgtype = record.pwnlib_msgtype
|
|
record.pwnlib_msgtype = 'animated'
|
|
msg = "%s\n" % self.format(record)
|
|
|
|
# we enrich the `Progress` object to keep track of the spinner
|
|
if not hasattr(progress, '_spinner_handle'):
|
|
spinner_handle = term.output('[x] ')
|
|
msg_handle = term.output(msg)
|
|
stop = threading.Event()
|
|
def spin():
|
|
'''Wheeeee!'''
|
|
state = 0
|
|
states = random.choice(spinners.spinners)
|
|
while True:
|
|
prefix = '[%s] ' % _spinner_style(states[state])
|
|
spinner_handle.update(prefix)
|
|
state = (state + 1) % len(states)
|
|
if stop.wait(0.1):
|
|
break
|
|
t = Thread(target = spin)
|
|
t.daemon = True
|
|
t.start()
|
|
progress._spinner_handle = spinner_handle
|
|
progress._msg_handle = msg_handle
|
|
progress._stop_event = stop
|
|
progress._spinner_thread = t
|
|
else:
|
|
progress._msg_handle.update(msg)
|
|
|
|
# if the message type was not a status message update, then we should
|
|
# stop the spinner
|
|
if msgtype != 'status':
|
|
progress._stop_event.set()
|
|
progress._spinner_thread.join()
|
|
style, symb = _msgtype_prefixes[msgtype]
|
|
prefix = '[%s] ' % style(symb)
|
|
progress._spinner_handle.update(prefix)
|
|
|
|
class Formatter(logging.Formatter):
|
|
"""
|
|
Logging formatter which performs custom formatting for log records
|
|
containing the ``'pwnlib_msgtype'`` attribute. Other records are formatted
|
|
using the `logging` modules default formatter.
|
|
|
|
If ``'pwnlib_msgtype'`` is set, it performs the following actions:
|
|
|
|
* A prefix looked up in `_msgtype_prefixes` is prepended to the message.
|
|
* The message is prefixed such that it starts on column four.
|
|
* If the message spans multiple lines they are split, and all subsequent
|
|
lines are indented.
|
|
|
|
This formatter is used by the handler installed on the ``'pwnlib'`` logger.
|
|
"""
|
|
|
|
# Indentation from the left side of the terminal.
|
|
# All log messages will be indented at list this far.
|
|
indent = ' '
|
|
|
|
# Newline, followed by an indent. Used to wrap multiple lines.
|
|
nlindent = '\n' + indent
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
# use the default formatter to actually format the record
|
|
msg = super(Formatter, self).format(record)
|
|
|
|
# then put on a prefix symbol according to the message type
|
|
|
|
msgtype: str | None = getattr(record, 'pwnlib_msgtype', None)
|
|
|
|
# if 'pwnlib_msgtype' is not set (or set to `None`) we just return the
|
|
# message as it is
|
|
if msgtype is None:
|
|
return msg
|
|
|
|
if msgtype in _msgtype_prefixes:
|
|
style, symb = _msgtype_prefixes[msgtype]
|
|
prefix = '[%s] ' % style(symb)
|
|
elif msgtype == 'indented':
|
|
prefix = self.indent
|
|
elif msgtype == 'animated':
|
|
# the handler will take care of updating the spinner, so we will
|
|
# not include it here
|
|
prefix = ''
|
|
else:
|
|
# this should never happen
|
|
prefix = '[?] '
|
|
|
|
msg = prefix + msg
|
|
msg = self.nlindent.join(msg.splitlines())
|
|
return msg
|
|
|
|
def _need_text(s: object) -> str:
|
|
# circular import wrapper
|
|
global _need_text
|
|
from pwnlib.util.packing import _need_text as real_need_text
|
|
_need_text = real_need_text
|
|
return real_need_text(s, 2)
|
|
|
|
# we keep a dictionary of loggers such that multiple calls to `getLogger` with
|
|
# the same name will return the same logger
|
|
def getLogger(name: str | None) -> Logger:
|
|
return Logger(logging.getLogger(name))
|
|
|
|
class LogfileHandler(logging.FileHandler):
|
|
def __init__(self):
|
|
super(LogfileHandler, self).__init__('', delay=True)
|
|
@property
|
|
def stream(self) -> TextIOWrapper:
|
|
return context.log_file
|
|
@stream.setter
|
|
def stream(self, value: TextIOWrapper) -> None:
|
|
pass
|
|
def handle(self, record: logging.LogRecord) -> bool:
|
|
if self.stream.name is not None:
|
|
return super(LogfileHandler, self).handle(record)
|
|
return False
|
|
|
|
iso_8601 = '%Y-%m-%dT%H:%M:%S'
|
|
fmt = '%(asctime)s:%(levelname)s:%(name)s:%(message)s'
|
|
log_file = LogfileHandler()
|
|
log_file.setFormatter(logging.Formatter(fmt, iso_8601))
|
|
|
|
#
|
|
# The root 'pwnlib' logger is declared here. To change the target of all
|
|
# 'pwntools'-specific logging, only this logger needs to be changed.
|
|
#
|
|
# Logging cascades upward through the hierarchy,
|
|
# so the only point that should ever need to be
|
|
# modified is the root 'pwnlib' logger.
|
|
#
|
|
# For example:
|
|
# map(rootlogger.removeHandler, rootlogger.handlers)
|
|
# logger.addHandler(myCoolPitchingHandler)
|
|
#
|
|
rootlogger = getLogger('pwnlib')
|
|
console = Handler()
|
|
formatter = Formatter()
|
|
console.setFormatter(formatter)
|
|
|
|
def install_default_handler() -> None:
|
|
'''install_default_handler()
|
|
|
|
Instantiates a :class:`Handler` and :class:`Formatter` and installs them for
|
|
the ``pwnlib`` root logger. This function is automatically called from when
|
|
importing :mod:`pwn`.
|
|
'''
|
|
logger = logging.getLogger('pwnlib')
|
|
|
|
if console not in logger.handlers:
|
|
logger.addHandler(console)
|
|
logger.addHandler(log_file)
|
|
|
|
logger.setLevel(1)
|