Module `tenlib.flow`

Main input/output functions.

Here's a base ten python file layout:

#!/usr/bin/env python

from ten import *


@entry
def main():
    ...


main()

The main coroutine's parameters are automatically mapped to a argparse argument (see entry()).

To display messages, use msg_* functions (msg_info() or msg_success() for instance).

If you script fails, you can stop the execution at any time using leave(), failure() or error(). Additionally, use assume() to ensure a condition is valid. If it is not, the script will exit.

Execution can be paused using pause(), or delayed using sleep().

To inform the user of progress in real time, use msg_status(), progress(), or track(), or use the inform() decorator.

Most of the heavy lifting for display is done by the rich library.

Sub-modules

tenlib.flow.console: Stores the console object used by ten.
tenlib.flow.messageformatter: This module provides different styles to display msg_* messages …

Functions

def entry(entrypoint: Callable[..., ~FRetType]) ‑> Callable[[], ~FRetType]

Converts the given coroutine or class into the program's entry point. Command line arguments are mapped to the parameters of the function or the object's __init__. Also, TenExit and KeyboardInterrupt exceptions will be caught and displayed nicely.

Default values, as well as type hinting, are supported.

The generic look of the program with a coroutine is:

#!/usr/bin/env python3
from ten import *

@entry
def main():
    ...

main()

And with a class:

#!/usr/bin/env python3
from ten import *

@entry
class Program:
    def __init__(self, ...):
        ...

    def run(self):
        ...

Program()

Examples

Creating a script with 1 required parameter, and 2 optional parameters:

#!/usr/bin/env python3

from ten import *

@entry
def exploit(url, username='admin', password='password'):
    """Exploits a post-authentication vulnerability on given URL.
    """
    msg_info(f'URL: {url}')
    msg_info(f'Username: {username}')
    msg_info(f'Password: {password}')
    ...

exploit()

Running the script:

$ /tmp/exploit.py -h
usage: boo.py [-h] [-u USERNAME] [-p PASSWORD] url

Exploits a post-authentication vulnerability on given URL.

positional arguments:
url

optional arguments:
-h, --help            show this help message and exit
-u USERNAME, --username USERNAME
-p PASSWORD, --password PASSWORD
$ /tmp/exploit.py 'http://target.com'
⊙ URL: <http://target.com>
⊙ Username: admin
⊙ Password: password
$ /tmp/exploit.py 'http://target.com' -u user1000
⊙ URL: <http://target.com>
⊙ Username: user1000
⊙ Password: password
$

Alternatively, with a class:

#!/usr/bin/env python3

from ten import *

@entry
class Exploit:
    def __init__(self, url, username='admin', password='password'):
        self.url = url
        self.username = username
        self.password = password

    def run(self):
        msg_info(f'URL: {self.url}')
        msg_info(f'Username: {self.username}')
        msg_info(f'Password: {self.password}')
        ...

Exploit()

Other supported functions declaration:

@entry
def exploit(url, table=None, fields=['user()', 'version()']):
    """Gets a URL, an SQL table (str) and a list of SQL fields.
    """
    ...

@entry
def sum_all(numeric_values: list[int]):
    """Takes a list of at least one int value as input.
    """
    msg_info(f'Sum: {sum(numeric_values)}')

@entry
def sum_all(numeric_values: [123.4, 567.8]):
    """Takes a list of at least one float value as input.
    """
    msg_info(f'Sum: {sum(numeric_values)}')

@entry
def add(a=3.12, b=4.12):
    """Adds two float numbers.
    """
    msg_success(f'Sum: {a + b}')

@entry
def force_opts(a, *, b, c):
    """Forces b and c to be optional arguments.
    """
    msg_success(f'Options: b={b!r} c={c!r}')

def arg(name: str, description: str) ‑> Callable[[~FCallable], ~FCallable]

Provides documentation for the parameter name of the entry() function. The documentation is visible when the program is run with the --help flag.

Example:

@entry
@arg("name", "The name of the person to greet")
def hello(name: str):
    msg_info(f"Hello, {name}!")

def msg_info(*objects, **kwargs)

Displays an info message.

def msg_failure(*objects, **kwargs)

Displays a failure message.

def msg_error(*objects, **kwargs)

Displays an error message.

def msg_success(*objects, **kwargs)

Displays a success message.

def msg_warning(*objects, **kwargs)

Displays a warning message.

def msg_print(*objects, **kwargs)

Displays a message.

def msg_debug(*objects, **kwargs)

Displays a debug message.

def msg_clear()

Clears the current line.

def msg_status(status: Union[rich.console.ConsoleRenderable, rich.console.RichCast, str], *, spinner: str = 'dots', spinner_style: str = 'status.spinner', speed: float = 1.0, refresh_per_second: float = 12.5) ‑> rich.status.Status

Display a status and spinner. See rich's Console.status.

Example

with msg_status("Doing something..."):
    # Do something slow
    ...

Args

status : RenderableType: A status renderable (str or Text typically).
spinner : str, optional: Name of spinner animation (see python -m rich.spinner). Defaults to "dots".
spinner_style : StyleType, optional: Style of spinner. Defaults to "status.spinner".
speed : float, optional: Speed factor for spinner animation. Defaults to 1.0.
refresh_per_second : float, optional: Number of refreshes per second. Defaults to 12.5.

Returns

Status: A Status object that may be used as a context manager.

def bin_print(data: bytes)

Prints bytes to stdout.

def leave(message=None, *args, **kwargs) ‑> NoReturn

Displays a message if given, and exits.

Raises

TenExit: to exit

def failure(message: str = None) ‑> NoReturn

Displays a failure message if given, and exits.

Raises

TenFailure: to exit

def error(message: str = None) ‑> NoReturn

Displays an error message if given, displays the stack trace, and exits.

Raises

TenError: to exit

def assume(assumption: bool, message: str = 'Assumption failed') ‑> None

If given assumption is False, raises TenFailure. This function is the equivalent of assert, but it'll display a standard failure message and exit, without displaying a stack trace.

In other words, if the assumption being false means the program has failed, but is not an error per-se, use assume(). If you want an exception to be raised, use the built-in assert.

This allows you to simplify code such as:

match = response.re.search(r'token="(.*?)"')
if not match:
    failure('Response does not contain a token')

to:

match = response.re.search(f'token="(.*?)"')
assume(match, 'Response does not contain a token')

Args

assumption : bool: Assumption to evaluate
message : str: Message for the TenFailure exception

def inform(go: str = None, ok: str = None, ko: str = None, ko_exit: bool = False) ‑> Callable[[~FCallable], ~FCallable]

This decorator will display a spinner and a message while a coroutine is running, and optionally a message after it is done.

Args

go : str: Message to display while the coro executes
ok : str: Message to display if the coro returns a result that evaluates to True.
ko : str: Message to display if the coro returns a result that evaluates to False.
ko_exit : bool: If set, the execution will stop if the coro does not find a result. Defaults to False.

ok and ko can use the {result} format to display the result.

Examples

Decorate a function that bruteforces a seed:

@inform(
    go='Bruteforcing seed...',
    ok='Seed found: {result}',
    ko='Unable to find seed'
)
def find_seed(token):
    for i in range(1000000):
        if hashing.md5(i) == token:
            return i
    return None

Decorate a login procedure: if it fails, the script will exit.

@inform(
    go='Trying to log in...',
    ok='Login successful',
    ko='Login failed, exiting',
    ko_exit=True
)
def login(session, user, password):
    r = await session.post('/login', {'u': user, 'p': password})
    return r.contains('login successful')

def pause(message: str = 'Paused. Press ENTER to resume execution') ‑> None

Pauses the execution of the program until ENTER is pressed.

Args

message : str: A message to display before pausing

def sleep(seconds) ‑> None

Alias for time.sleep.

def progress(transient: bool = True, **kwargs) ‑> rich.progress.Progress

A Progress bar.

Args

transient : bool: Clear the progress on exit. Defaults to False.
**kwargs: see the documentation for Progress.

def track(sequence: Iterable[~FIterableItem], description: str = 'Working...', total: Optional[float] = None, transient: bool = True) ‑> Generator[~FIterableItem, None, None]

Track progress when iterating over a sequence. This is a wrapper for rich's track.

Example

for i in track(range(100), "Iterating over 100 numbers"):
    # do something
    ...

Args

sequence : Iterable: A sequence (must support len()) you wish to iterate over.
description : str, optional: Description of task show next to progress bar. Defaults to "Working".
total : float, optional: Total number of steps. Default is len(sequence).
transient : bool, optional: Clear the progress on exit. Defaults to True.

Returns

Iterable: The original iterable.

def ask(prompt: Union[str, ForwardRef('Text')] = '', *, password: bool = False, choices: Optional[list[str]] = None, show_default: bool = True, show_choices: bool = True, default: Any = Ellipsis) ‑> Any

Shortcut to construct and run a prompt loop and return the result. Wrapper for rich.prompt.Prompt.

Example

>>> filename = ask("Enter a filename")

Args

prompt : TextType, optional: Prompt text. Defaults to "".
password : bool, optional: Enable password input. Defaults to False.
choices : list[str], optional: A list of valid choices. Defaults to None.
show_default : bool, optional: Show default in prompt. Defaults to True.
show_choices : bool, optional: Show choices in prompt. Defaults to True.

def trace(function: ~FCallable) ‑> ~FCallable

Decorator that displays a debug log line containing the function name, its parameters and the value returned.

def set_message_formatter(message_formatter: Union[MessageFormatter, Type[MessageFormatter], str]) ‑> None

Sets given MessageFormatter as the message formatter.

def set_random_message_formatter() ‑> None

Sets a random MessageFormatter as the message formatter.