Module reference

Client for managing Sieve scripts remotely using the ManageSieve protocol

exception sievemgr.AppConfigError[source]

Bases: AppError, ConfigError

Applicaiton configuration error.

exception sievemgr.AppConnectionError[source]

Bases: AppError, ConnectionError

Client-side connection error.

exception sievemgr.AppError[source]

Bases: Error

Base class for application errors.

exception sievemgr.AppOperationError[source]

Bases: AppError, OperationError

Client-side operation error.

exception sievemgr.AppSecurityError[source]

Bases: AppError, SecurityError

Client security error.

exception sievemgr.AppSoftwareError[source]

Bases: AppError, SoftwareError

Client software error (aka a bug).

exception sievemgr.CapabilityError[source]

Bases: Error

Base class for capability errors.

exception sievemgr.ConfigError[source]

Bases: Error

Base class for configuration errors.

exception sievemgr.DNSDataError[source]

Bases: Error

DNS data error.

exception sievemgr.DNSError[source]

Bases: Error

Base class for DNS errors.

exception sievemgr.DNSOperationError[source]

Bases: DNSError, OperationError

DNS operation error.

exception sievemgr.DNSSoftwareError[source]

Bases: DNSError, SoftwareError

DNS software error.

exception sievemgr.DataError[source]

Bases: Error

Base class for data errors.

exception sievemgr.Error[source]

Bases: Exception

Base class for errors.

exception sievemgr.HTTPError[source]

Bases: Error

Base class for HTTP errors.

exception sievemgr.HTTPOperationError[source]

Bases: HTTPError, OperationError

HTTP operation error.

exception sievemgr.HTTPUsageError[source]

Bases: HTTPError, ProtocolError

HTTP usage error.

exception sievemgr.OCSPDataError[source]

Bases: OCSPError, DataError

OCSP data error.

exception sievemgr.OCSPError[source]

Bases: Error

Base class for OCSP errors.

exception sievemgr.OCSPOperationError[source]

Bases: OCSPError, OperationError

OCSP operation error.

exception sievemgr.OperationError[source]

Bases: Error

Base class for operation errors.

exception sievemgr.ProtocolError[source]

Bases: Error

Base class for protocol errors.

Danger

Continuing after a ProtocolError may cause undefined behaviour.

exception sievemgr.SASLCapabilityError[source]

Bases: Error

SASL capability error.

exception sievemgr.SASLError[source]

Bases: Error

Base class for SASL errors.

exception sievemgr.SASLProtocolError[source]

Bases: SASLError, ProtocolError

Server violated the SASL protocol.

exception sievemgr.SASLSecurityError[source]

Bases: SASLError, SecurityError

SASL security error.

exception sievemgr.SecurityError[source]

Bases: Error

Base class for security errors.

Danger

Continuing after a SecurityError compromises the connection.

exception sievemgr.ShellError[source]

Bases: Error

Base class for shell errors.

exception sievemgr.ShellOperationError[source]

Bases: ShellError, OperationError

Shell operation error.

exception sievemgr.ShellUsageError[source]

Bases: ShellError, UsageError

Shell usage error.

exception sievemgr.SieveCapabilityError[source]

Bases: SieveError, CapabilityError

Capability not supported by the server.

exception sievemgr.SieveConnectionError[source]

Bases: Response, SieveError, ConnectionError

Server said “BYE”.

__init__(response: Atom = Atom('BYE'), code: tuple[Word, ...] = (), message: str | None = None)[source]
exception sievemgr.SieveError[source]

Bases: Error

Base class for ManageSieve errors.

exception sievemgr.SieveOperationError[source]

Bases: Response, SieveError, OperationError

Server said “NO”.

__init__(response: Atom = Atom('NO'), code: tuple[Word, ...] = (), message: str | None = None)[source]
exception sievemgr.SieveProtocolError[source]

Bases: SieveError, ProtocolError

Server violated the ManageSieve protocol error.

exception sievemgr.SignalCaught[source]

Bases: Exception

Signal was caught.

__init__(signo: int, frame: FrameType | None = None) None
classmethod catch(*signals: int) Callable[[Callable[[...], T]], Callable[[...], T]][source]

Decorator that handles signals.

If one of the given signals is caught, SignalCaught is raised. If that exception is not caught, the process group is terminated, the signal handler reset, and the signal re-raised.

classmethod register(signals: Iterable[int]) tuple[Callable[[int, FrameType | None], Any] | int | None, ...][source]

Register throw() as handler for the given signals.

Parameters:

signals – Signals to register throw() as handler for.

Returns:

Old signal handlers.

classmethod throw(signo: int, frame: FrameType | None)[source]

Raise a SignalCaught exception.

frame: FrameType | None = None

Stack frame.

signo: int

Signal number.

exception sievemgr.SoftwareError[source]

Bases: Error

Base class for software errors.

exception sievemgr.TLSCapabilityError[source]

Bases: TLSError, CapabilityError

TLS capability error.

exception sievemgr.TLSError[source]

Bases: Error

Base class for TLS errors.

exception sievemgr.TLSSecurityError[source]

Bases: TLSError, OperationError

TLS security error.

exception sievemgr.TLSSoftwareError[source]

Bases: TLSError, SoftwareError

TLS software error.

exception sievemgr.UsageError[source]

Bases: Error

Base class for usage errors.

class sievemgr.AbstractAuth[source]

Bases: ABC

Abstract base class for SASL mechanisms.

The ManageSieve “AUTHENTICATE” command performs a Simple Authentication and Security Layer (SASL) protocol exchange. SASL is a framework that comprises different authentication mechanisms (“SASL mechanisms”). SieveManager.authenticate() does not implement any such mechanism itself, but delegates the SASL protocol exchange to classes that do. Such classes must subclass this class and have a name attribute that indicates the mechanism they implemented.

Tip

Do not subclass AbstractAuth directly. Subclass BaseAuth instead.

See also

AbstractSASLAdapter

Abstract base class for sending and receiving SASL messages.

RFC 3454

Preparation of Internationalized Strings

RFC 4013

Stringprep Profile for User Names and Passwords

RFC 4422

Simple Authentication and Security Layer (SASL)

RFC 5804 (sec. 2.1)

ManageSieve “AUTHENTICATE” command

abstract __init__(conn: AbstractSASLAdapter, authcid: str, authzid: str = '', prepare: SASLPrep = SASLPrep.ALL)[source]

Prepare authentication.

authcid and authzid are prepared according to RFC 3454 and RFC 4013 if the specification of the SASL mechanism mandates or recommends string preparation and prepare & SASLPrep.USERNAMES evaluates as true.

Parameters:
  • conn – Connection over which to authenticate.

  • authcid – Authentication ID (user to login as).

  • authzid – Authorisation ID (user whose rights to acquire).

  • prepare – Which credentials to prepare.

Raises:

ValueError – Bad characters in username.

classmethod getmechs(sort: bool = True, obsolete: bool = False) list[type[AbstractAuthT]][source]

Get authentication classes that subclass this class.

Parameters:
  • sort – Sort mechanisms by order?

  • obsolete – Return obsolete mechanisms?

authcid: str

Authentication ID (user to login as).

authzid: str = ''

Authorisation ID (user whose rights to acquire).

name: ClassVar[str]

Mechanism name.

obsolete: bool = False

Is this mechanism obsolete?

order: int = 0

Mechanism precedence.

class sievemgr.AbstractAuthT

Type variable for AbstractAuth.

alias of TypeVar(‘AbstractAuthT’, bound=AbstractAuth)

class sievemgr.AbstractSASLAdapter[source]

Bases: ABC

Abstract base class for sending and receiving SASL messages.

Messages that comprise an SASL protocol exchange (“SASL messages”) must be translated to the underlying protocol. This class defines the types of messages that may occur in an SASL protocol exchange. Classes that translate between SASL and the underlying protocol must subclass this class.

See also

AbstractAuth

Abstract base class for SASL mechanisms.

RFC 4422

Simple Authentication and Security Layer (SASL)

abstract abort()[source]

Abort authentication.

Raises:

ProtocolError – Protocol violation.

abstract begin(name: str, data: bytes | None = None)[source]

Begin authentication.

Parameters:
  • name – SASL mechanism name.

  • data – Optional client-first message.

Raises:
abstract end()[source]

Conclude authentication.

Raises:
abstract receive() bytes[source]

Receive and decode an SASL message.

Raises:
abstract send(data: bytes)[source]

Encode and send an SASL message.

Raises:
abstract property sock: socket | SSLSocket

Underlying socket.

class sievemgr.Atom[source]

Bases: str

ManageSieve keyword (e.g., LISTSCRIPTS, OK).

enum sievemgr.AuthState(value)[source]

Bases: IntEnum

State of the authentication process.

Member Type:

int

Valid values are as follows:

PREAUTH = <AuthState.PREAUTH: 1>
SENT = <AuthState.SENT: 2>
RECEIVED = <AuthState.RECEIVED: 3>
DONE = <AuthState.DONE: 4>

The Enum and its members also have the following methods:

__new__(value)
class sievemgr.AuthzUnsupportedMixin[source]

Bases: object

Mixin for SASL mechanisms that do not support authorisation.

For example:

__init__(*args, **kwargs)[source]

Prepare authentication.

Raises:

SASLCapabilityErrorauthzid is set.

class sievemgr.BaseACAPConn[source]

Bases: ABC

Base class for ACAP parsers/serialisers.

The ManageSieve uses the syntax and data types of the Application Control Access Protocol (ACAP). This class provides a parser for converting ACAP lines into Python objects and a serialiser for converting Python objects into ACAP lines.

See also

RFC 2244 (secs. 2.2 and 2.6)

ACAP commands, responses, and data formats.

RFC 5804 (secs. 1.2 and 4)

ManageSieve syntax.

receiveline() Line[source]

Receive a line and parse it.

ACAP type

Python type

Atom

Atom

Literal

str

Nil

None

Number

int

Parenthesised List

list

String

str

For example:

>>> mgr.sendline(Atom('listscripts'))
>>> mgr.receiveline()
['foo.sieve', 'ACTIVE']
>>> mgr.receiveline()
['bar.sieve']
>>> mgr.receiveline()
['baz.sieve']
>>> mgr.receiveline()
['OK', 'Listscripts completed.']
Raises:

ValueError – Line is malformed.

sendline(*objs: IO[Any] | Word, whole: bool = True)[source]

Convert objs to ACAP types and send them.

Python type

ACAP type

Atom

Atom

typing.IO

Literal

None

Nil

bytes

Literal or String [1]

list

Parenthesised List

int

Number [2]

str

Literal or String [1] [3]

For example:

>>> mgr.sendline(Atom('havespace'), 'script.sieve', 12345)
>>> mgr.receiveline()
['OK', 'Putscript would succeed.']

Pipeline commands:

>>> mgr.isalive(check=True)
>>> with open('foo.sieve') as script:
>>>     mgr.sendline(Atom('putscript', script, 'foo.sieve'))
>>> mgr.sendline(Atom('logout'))
>>> for _ in range(2):
>>>     mgr.collect(check=True)
Parameters:
  • objs – Objects to serialise.

  • whole – Conclude data with CRLF?

Raises:
  • ValueError – Number is not within the range [0, 4,294,967,295].

  • TypeError – Object cannot be represented as ACAP data type.

abstract property file: IO | BufferedRWPair | LogIOWrapper | None

File-like access to the underlying socket.

class sievemgr.BaseAuth[source]

Bases: AbstractAuth, ABC

Base class for authentication mechanisms.

BaseAuth provides methods to prepare strings according to RFC 3454 and RFC 4013 and a layer over the underlying AbstractSASLAdapter object that calls AbstractSASLAdapter.begin() and AbstractSASLAdapter.end() transparently.

Credentials must be prepared in __init__(). Subclasses should pass connection, authcid, authzid, and prepare to super().__init__ and use prepare() to prepare the remaining credentials. For example:

def __init__(self, connection: AbstractSASLAdapter,
             authcid: str, password: str, authzid: str = '',
             prepare: SASLPrep = SASLPrep.ALL):
    """Prepare authentication.

    `authcid`, `password`, and `authzid` are prepared according to
    :rfc:`3454` and :rfc:`4013` if ``prepare & SASLPrep.USERNAMES``
    and/or ``prepare & SASLPrep.PASSWORDS`` are non-zero.

    Arguments:
        conn: Connection over which to authenticate.
        authcid: Authentication ID (user to login as).
        password: Password.
        authzid: Authorisation ID (user whose rights to acquire).
        prepare: Which credentials to prepare.

    Raises:
        ValueError: Bad characters in username or password.
    """
    super().__init__(connection, authcid, authzid, prepare)
    prepare &= SASLPrep.PASSWORDS   # type: ignore[assignment]
    self.password = self.prepare(password) if prepare else password

The SASL exchange must be implemented in exchange(). Subclasses should use send() and receive() to exchange SASL messages. For example:

def exchange(self):
    data = '\0'.join((self.authzid, self.authcid, self.password))
    self.send(data.encode('utf8'))
__init__(adapter: AbstractSASLAdapter, authcid: str, authzid: str = '', prepare: SASLPrep = SASLPrep.ALL)[source]

Prepare authentication.

authcid and authzid are prepared according to RFC 3454 and RFC 4013 if prepare & SASLPrep.USERNAMES evaluates as true.

Parameters:
  • conn – Connection over which to authenticate.

  • authcid – Authentication ID (user to login as).

  • authzid – Authorisation ID (user whose rights to acquire).

  • prepare – Which credentials to prepare.

Raises:

ValueError – Bad characters in username.

abort()[source]

Abort authentication.

Raises:

ProtocolError – Protocol violation.

begin(data: bytes | None = None)[source]

Begin authentication.

Parameters:

data – Optional client-first message.

Raises:
end()[source]

Conclude authentication.

Raises:
abstract exchange()[source]

Exchange SASL messages.

static prepare(string: str) str[source]

Prepare string according to RFC 3454 and RFC 4013.

Returns:

Prepared string.

Raises:

ValueErrorString is malformed.

receive() bytes[source]

Receive and decode an SASL message.

Raises:

Note

Calls begin() if needed.

send(data: bytes)[source]

Encode and send an SASL message.

Raises:

Note

Calls begin() if needed.

adapter: AbstractSASLAdapter

Underlying SASL adapter.

property sock: socket | SSLSocket

Underlying socket.

state: AuthState = 1

Current authentication state.

class sievemgr.BaseConfig[source]

Bases: UserDict

Base class for configurations.

loadfile(fname: str)[source]

Read configuration variables from fname.

Raises:

AppConfigError – Syntax error.

parse(expr: str)[source]

Split expr into a name and a value and set the variable.

Expr is split at the first equals sign (“=”). var is equivalent to var=yes. novar is equivalent to var=no.

set(name: str, value)[source]

Set the configuration variable name to value.

Raises:

AttributeError – Bad variable.

property sections: dict[str, BaseConfigT]

Sections in the loaded configuration files.

class sievemgr.BaseConfigT

Type variable for BaseConfig.

alias of TypeVar(‘BaseConfigT’, bound=BaseConfig)

class sievemgr.BasePwdAuth[source]

Bases: BaseAuth, ABC

Base class for password-based authentication mechanisms.

Prepares credentials, so that subclasses need only implement exchange(). For example:

class PlainAuth(BasePwdAuth):
    """PLAIN authentication.

    .. seealso::
        :rfc:`4616`
            PLAIN authentication mechanism.
    """

    def exchange(self):
        data = '\0'.join((self.authzid, self.authcid, self.password))
        self.send(data.encode('utf8'))

    name = 'PLAIN'
__init__(connection: AbstractSASLAdapter, authcid: str, password: str, authzid: str = '', prepare: SASLPrep = SASLPrep.ALL)[source]

Prepare authentication.

authcid, password, and authzid are prepared according to RFC 3454 and RFC 4013 if prepare & SASLPrep.USERNAMES and/or prepare & SASLPrep.PASSWORDS are non-zero.

Parameters:
  • conn – Connection over which to authenticate.

  • authcid – Authentication ID (user to login as).

  • password – Password.

  • authzid – Authorisation ID (user whose rights to acquire).

  • prepare – Which credentials to prepare.

Raises:

ValueError – Bad characters in username or password.

password: str

Password.

class sievemgr.BaseScramAuth[source]

Bases: BasePwdAuth, ABC

Base class for SCRAM authentication mechanisms.

Implements exchange(), so that subclasses need only define a digest. For example:

class ScramSHA1Auth(BaseScramAuth):
    """SCRAM-SHA-1 authentication."""

    @property
    def digest(self) -> str:
        return 'sha1'

    name = 'SCRAM-SHA-1'
    order = -10

See also

RFC 5802

Salted Challenge Response Authentication Mechanism (SCRAM).

RFC 7677

SCRAM-SHA-256 and SCRAM-SHA-256-PLUS.

https://datatracker.ietf.org/doc/html/draft-melnikov-scram-bis

Updated recommendations for implementing SCRAM.

https://datatracker.ietf.org/doc/html/draft-melnikov-scram-sha-512-03

SCRAM-SHA-512 and SCRAM-SHA-512-PLUS.

https://datatracker.ietf.org/doc/html/draft-melnikov-scram-sha3-512-03

SCRAM-SHA3-512 and SCRAM-SHA3-512-PLUS.

https://csb.stevekerrison.com/post/2022-01-channel-binding

Discussion of TLS channel binding.

https://csb.stevekerrison.com/post/2022-05-scram-detail

Discussion of SCRAM.

exchange()[source]

Exchange SASL messages.

cbdata: bytes = b''

TLS channel-binding data.

cbtype: str = ''

TLS channel-binding type.

abstract property digest: str

Digest name as used by hashlib and hmac.

noncelen: int = 18

Client nonce length in bytes.

class sievemgr.BaseScramPlusAuth[source]

Bases: BaseScramAuth, ABC

Base class for SCRAM mechanisms with channel binding.

For example:

class ScramSHA1PlusAuth(BaseScramPlusAuth, ScramSHA1Auth):
    """SCRAM-SHA-1-PLUS authentication."""
    name = 'SCRAM-SHA-1-PLUS'
    order = -1000
__init__(*args, **kwargs)[source]

Prepare authentication.

authcid, password, and authzid are prepared according to RFC 3454 and RFC 4013 if prepare & SASLPrep.USERNAMES and/or prepare & SASLPrep.PASSWORDS are non-zero.

Parameters:
  • conn – Connection over which to authenticate.

  • authcid – Authentication ID (user to login as).

  • password – Password.

  • authzid – Authorisation ID (user whose rights to acquire).

  • prepare – Which credentials to prepare.

Raises:

ValueError – Bad characters in username or password.

class sievemgr.BaseShell[source]

Bases: object

Base class for interactive shells.

BaseShell is similar-ish to cmd.Cmd. However, lines read from standard input are expanded before being passed to methods that implement commands. BaseShell also provides an extensible completion system as well as a built-in help system. aliases can be extended, too.

Define a do_command method to add command.

For example:

>>> class Calculator(BaseShell):
>>>     def do_add(self, n, m):
>>>         """add n m - add n and m"""
>>>         print(n + m)
>>>
>>>     def do_sum(self, *numbers):
>>>         """sum [n ...] - """
>>>         print sum(numbers)
>>>
>>>     aliases = {
>>>         '+': 'sum'
>>>     }
>>>
>>> calc = Calculator()
>>> calc.executeline('add 0 1')
1
>>> calc.executeline('sum 0 1 2')
3
>>> calc.executeline('+ 0 1 2 3')
6
>>> calc.executeline('add 0')
usage: add n m
>>> calc.executeline('help add')
add n m - add n and m
__init__()[source]

Initialise a BaseShell object.

static columnise(words: Sequence[str], file: TextIO = sys.stdout, width: int = shutil.get_terminal_size().columns)[source]

Print words in columns to file.

Parameters:
  • words – Words.

  • file – Output file.

  • width – Terminal width in chars.

complete(text: str, n: int) str | None[source]

Completion function for readline.set_completer().

Tab-completion for command names (e.g., “exit”) is built in. To enable tab-completion for the arguments of some command, define a method complete_command, which takes the index of the argument that should be completed and the given text and returns a sequence of str-bool pairs, where the string is a completion and the boolean indicates whether a space should be appended to that completion.

For example:

>>> class Foo(BaseShell)
>>>     def do_cmd(self, arg1, arg2):
>>>         pass
>>>
>>>     def complete_cmd(self, argidx, text):
>>>         if argidx == 1:
>>>             return [(s, True) for s in ('foo', 'bar', 'baz')]
>>>         if argidx == 2:
>>>             return [('quux/', False)]
>>>         return []
>>>
>>> foo = Foo()
>>> foo.complete('', 0)
'cmd '
>>> foo.complete('', 1)
None
>>> foo.complete('cmd ', 0)
'foo '
>>> foo.complete('cmd ', 1)
'bar '
>>> foo.complete('cmd ', 2)
'baz '
>>> foo.complete('cmd ', 3)
None
>>> foo.complete('cmd b', 0)
'bar '
>>> foo.complete('cmd b', 1)
'baz '
>>> foo.complete('cmd b', 2)
None
>>> foo.complete('cmd bar ', 0)
'quux/'
>>> foo.complete('cmd bar ', 1)
None
Parameters:
  • text – Possibly partial word to be completed.

  • n – Index of the completion to return.

Returns:

Either the n-th completion for text or None if there is no such completion.

Side-effects

  • Logging of messages with a priority lower than logging.ERROR is suppressed during tab-completion.

  • A BEL is printed to the controlling terminal if no completion for text is found.

complete_help(*_) tuple[tuple[str, bool], ...][source]

Completer for help.

static confirm(prompt: str, default: ConfirmEnum = ConfirmEnum.NO, multi: bool = False, attempts: int = 3) ConfirmEnum[source]

Prompt the user for confirmation.

Parameters:
  • prompt – Prompt.

  • default – Default.

  • multi – Give choices “all” and “none”?

  • attempts – How often to try before raising a ValueError.

Raises:

ValueError – Unrecognised answer.

do_exit()[source]

exit - exit the shell

do_help(name: str | None = None)[source]

help [command] - list commands/show help for command

enter() int[source]

Start reading commands from standard input.

Reading stops at the end-of-file marker or when a command raises StopIteration.

Side-effects

Entering the shell binds Tab to readline.complete().

execute(command: str, *args: str) int[source]

Execute command.

For example:

>>> shell.execute('ls', 'foo', 'bar')
0
Parameters:
  • command – Command name.

  • args – Arguments to the command.

Returns:

Return value.

Raises:

ShellUsageError – Command not found or arguments invalid.

executeline(line: str) int[source]

Split line and execute() it.

For example:

>>> shell.executeline('ls foo bar')
0
Returns:

Return value.

Raises:

ShellUsageError – Command not found or arguments invalid.

executescript(script: TextIO) int[source]

Split script into lines and execute them.

For example:

>>> with open('scriptfile') as scriptfile:
>>>     shell.executescript(scriptfile)
0
Returns:

Return value.

Raises:

ShellUsageError – Command not found or arguments invalid.

expand(line: str) list[str][source]

Expand the words that comprise line.

Similar to wordexp(3). Patterns are expanded using fnmatch.fnmatchcase(), with filenames being provided by the completion system.

For example:

>>> class Foo(BaseShell)
>>>     def complete_cmd(self, _, text):
>>>         return [(s, True) for s in ('foo', 'bar', 'baz')]
>>>
>>> foo = Foo()
>>> foo.expand('cmd *')
['cmd', 'foo', 'bar', 'baz']
classmethod getargs() list[ShellPattern | str][source]

Split line before current completion scope.

classmethod getcommands() Iterator[str][source]

Get the shell commands provided by cls.

For example:

>>> class Foo(BaseShell)
>>>     def do_cmd(self, arg1, arg2):
>>>         pass
>>>
>>> tuple(Foo.getcommands())
('cmd', 'exit', 'help')
>>> foo = Foo()
>>> foo.commands
('cmd', 'exit', 'help')
getprompt() str[source]

Get a shell prompt.

static getusage(func: Callable) str | None[source]

Derive a usage message from func’s docstring.

getusage() assumes that the docstring has the form command args - description. The usage message is either the text up to the last dash (“-”) or, if the docstring does not contain a dash, the whole docstring. Leading and trailing whitespace is stripped.

For example:

>>> def frobnicate(foo, bar):
>>>     """frobnicate foo bar - frobnicate foo with bar"""
>>>     ...
>>>
>>> BaseShell.getusage(frobnicate)
'frobnicate foo bar'
static hasatty() bool[source]

Is standard input a terminal?

static split(line: str) list[ShellPattern | str][source]

Split line into words as a POSIX-compliant shell would.

aliases: dict[str, str] = {'!': 'sh', '?': 'help'}

Mapping of aliases to commands.

commands: tuple[str, ...]

Shell commands. Populated by __init__().

logger: Logger = <Logger sievemgr (WARNING)>

Logger.

Messages are logged with the following priorities:

Priority

Used for

logging.INFO

Help message when the shell is entered

retval: int = 0

Return value of the most recently completed command.

class sievemgr.BaseVar[source]

Bases: ABC

Base class for BaseConfig attributes.

For example:

>>> @dataclasses.dataclass
>>> class FooConfig(BaseConfig):
>>>     foo = BoolVar(default=False)
>>>     bar = NumVar(cls=int, default=0)
>>>
>>> foo = FooConfig(bar=1)
>>> foo.foo
False
>>> foo.bar
1
>>> foo.foo = 'yes'
>>> foo.foo
True
>>> foo.bar = '2'
>>> foo.bar
2
__init__(default: Any = None)[source]

Initialise a configuration variable.

default: Any

Default value.

name: str

Variable name.

class sievemgr.BoolVar[source]

Bases: BaseVar

Convert “yes” and “no” to bool.

class sievemgr.Capabilities[source]

Bases: object

Server capabilities.

__init__(implementation: str | None = None, sieve: tuple[str, ...] = (), language: str | None = None, maxredirects: int | None = None, notify: tuple[str, ...] = (), owner: str = '', sasl: tuple[str, ...] = (), starttls: bool = False, unauthenticate: bool = False, version: str | None = None, notunderstood: dict = <factory>) None
classmethod fromlines(lines: Iterable[Line]) CapabilitiesT[source]

Create a Capabilities object from a server response.

implementation: str | None = None

Server application.

language: str | None = None

Language.

maxredirects: int | None = None

Maximum number of redirect operations permitted in a script.

notify: tuple[str, ...] = ()

URI schema parts for supported notification methods.

notunderstood: dict

Capabilities not understood by SieveManager.

owner: str = ''

Canonical name of the user whose scripts are managed.

sasl: tuple[str, ...] = ()

Supported authentication methods.

sieve: tuple[str, ...] = ()

Supported Sieve modules.

starttls: bool = False

Is “STARTTLS” available?

unauthenticate: bool = False

Is “UNAUTHENTICATE” available?

version: str | None = None

ManageSieve protocol version.

class sievemgr.CapabilitiesT

Capabilities type variable.

alias of TypeVar(‘CapabilitiesT’, bound=Capabilities)

class sievemgr.CmdVar[source]

Bases: BaseVar, ExpandingVarMixin

Split up value into a list using shlex.split().

enum sievemgr.ConfirmEnum(value)[source]

Bases: IntEnum

Answers that BaseShell.confirm() may return.

Member Type:

int

Valid values are as follows:

NO = <ConfirmEnum.NO: 0>
YES = <ConfirmEnum.YES: 1>
ALL = <ConfirmEnum.ALL: 2>
NONE = <ConfirmEnum.NONE: 3>

The Enum and its members also have the following methods:

__new__(value)
class sievemgr.CramMD5Auth[source]

Bases: AuthzUnsupportedMixin, BasePwdAuth

CRAM-MD5 authentication.

See also

RFC 2195 (sec. 2)

Definition of CRAM-MD5.

exchange()[source]

Exchange SASL messages.

name: ClassVar[str] = 'CRAM-MD5'

Mechanism name.

obsolete: bool = True

Is this mechanism obsolete?

class sievemgr.EnumVar[source]

Bases: BaseVar

Convert comma-separated values to an enum.Enum.

__init__(*args, cls: type[Enum], **kwargs)[source]

Initialise the variable.

Parameters:
  • name – Variable name.

  • cls – Enumeration type.

  • default – Default value.

class sievemgr.ExpandingVarMixin[source]

Bases: object

Mixin for variables that do word expansion.

expand(obj: BaseConfig, value: str) str[source]

Expand ‘~’ and configuration variables.

class sievemgr.ExternalAuth[source]

Bases: BaseAuth

EXTERNAL authentication.

See also

RFC 4422 (App. A)

Definition of the EXTERNAL mechanism.

exchange()[source]

No-op.

name: ClassVar[str] = 'EXTERNAL'

Mechanism name.

class sievemgr.FilenameVar[source]

Bases: BaseVar

Expand ~user and make filenames absolute.

class sievemgr.FlagVar[source]

Bases: BaseVar, ListVarMixin

Convert comma-separated values to an int.

__init__(*args, cls: type[IntEnum], **kwargs)[source]

Initialise a configuration variable.

cls: type[IntEnum]

Enumeration type

class sievemgr.HostVar[source]

Bases: BaseVar

Check whether value is a valid hostname.

class sievemgr.ListVarMixin[source]

Bases: object

Mixin for lists.

splititems(maxsplit=0)

Split a comma-separated list into items.

class sievemgr.LogIOWrapper[source]

Bases: object

Logger for file-like objects.

__init__(file: BinaryIO | BufferedRWPair, encoding: str = 'utf8', level: int = logging.DEBUG, logger: Logger = logging.getLogger(__name__), formats: tuple[str, str] = ('S: %s', 'C: %s'))[source]

Log I/O to file.

Parameters:
  • file – File-like object opened in binary mode.

  • encodingfile’s encoding.

  • level – Logging priority.

  • logger – Logger.

  • formats – Message formats; ‘%s’ is replaced with I/O.

log(line: bytearray | bytes, fmt: str)[source]

Log line with fmt.

classmethod wrap(file: BinaryIO | BufferedRWPair, logger: Logger = logging.getLogger(__name__), level: int = logging.DEBUG, formats: tuple[str, str] = ('S: %s', 'C: %s'), encoding: str = 'utf8') BinaryIO | BufferedRWPair | LogIOWrapperT[source]

Wrap a file in a LogIOWrapper if logging is enabled.

Takes the same arguments as __init__().

Returns:

The file or a LogIOWrapper that wraps the file.

buffers: tuple[bytearray, bytearray]

Logging buffers.

encoding: str

file’s encoding.

file: BinaryIO | BufferedRWPair

Underlying file-like object.

formats: tuple[str, str]

Logging formats.

level: int

Logging level.

logger: Logger

Logger.

quiet: bool = False

Log I/O?

read: Callable[[...], bytes]

Read from file.

readinto: Callable[[...], int]

Read from file into a buffer.

readline: Callable[[...], bytes]

Read a line from file.

readlines: Callable[[...], list[bytes]]

Read all lines from file.

write: Callable[[...], int]

Write to file.

writelines: Callable[[...], None]

Write lines to file.

class sievemgr.LogIOWrapperT

Type variable for LogIOWrapper.

alias of TypeVar(‘LogIOWrapperT’, bound=LogIOWrapper)

enum sievemgr.LogLevel(value)[source]

Bases: IntEnum

Logging levels supported by SieveConfig.

Member Type:

int

Valid values are as follows:

AUTH = <LogLevel.AUTH: 5>
DEBUG = <LogLevel.DEBUG: 10>
INFO = <LogLevel.INFO: 20>
WARNING = <LogLevel.WARNING: 30>
ERROR = <LogLevel.ERROR: 40>

The Enum and its members also have the following methods:

__new__(value)
fromdelta(delta: int) LogLevel[source]

Get a LogLevel from a delta.

For example:

>>> LogLevel.INFO.fromdelta(-1)
<LogLevel.WARNING: 30>
>>> LogLevel.INFO.fromdelta(0)
<LogLevel.INFO: 20>
>>> LogLevel.INFO.fromdelta(1)
<LogLevel.DEBUG: 10>
>>> # Out-of-bounds deltas do not raise an error
>>> LogLevel.INFO.fromdelta(-math.inf)
<LogLevel.ERROR: 40>
>>> LogLevel.INFO.fromdelta(math.inf)
<LogLevel.AUTH: 5>
class sievemgr.LoginAuth[source]

Bases: AuthzUnsupportedMixin, BasePwdAuth

LOGIN authentication.

See also

https://datatracker.ietf.org/doc/draft-murchison-sasl-login

Definition of the LOGIN mechanism.

__init__(*args, **kwargs)[source]

Prepare authentication.

Parameters:
  • conn – Connection over which to authenticate.

  • authcid – Authentication ID (user to login as).

  • password – Password.

  • authzid – Authorisation ID (user whose rights to acquire).

  • prepare – Which credentials to prepare.

Raises:

ValueError – Password contains CR, LF, or NUL.

exchange()[source]

Exchange SASL messages.

name: ClassVar[str] = 'LOGIN'

Mechanism name.

obsolete: bool = True

Is this mechanism obsolete?

class sievemgr.NumVar[source]

Bases: BaseVar

Convert value to a number of type cls.

__init__(*args, cls: type = int, minval: float | int | None = None, maxval: float | int | None = None, **kwargs)[source]

Initialise the variable.

Parameters:
  • name – Variable name.

  • cls – Number type.

  • minval – Smallest permissible value.

  • maxval – Greatest permissible value.

  • default – Default value.

cls: type

Number type.

maxval: float | int | None

Maximum value.

minval: float | int | None

Minimum value.

class sievemgr.ObjWrapper[source]

Bases: dict

Object wrapper for use with code.interact().

Parameters:

obj – Object to wrap.

__init__(obj: Any)[source]

Initialise a proxy.

static exit()[source]

Exit the Python read-evaluate-print loop.

static help(*args, **kwargs)[source]

Show help.

class sievemgr.PlainAuth[source]

Bases: BasePwdAuth

PLAIN authentication.

See also

RFC 4616

PLAIN authentication mechanism.

exchange()[source]

Exchange SASL messages.

name: ClassVar[str] = 'PLAIN'

Mechanism name.

class sievemgr.Response[source]

Bases: object

Server response to a command.

See also

RFC 5804 (secs. 1.2, 1.3, 4, 6.4, and passim)

ManageSieve responses

__init__(response: Atom, code: tuple[Word, ...] = (), message: str | None = None) None
classmethod fromline(line: Line) ResponseT[source]

Create a Response object from a Line.

matches(*categories: str) bool[source]

Check if code matches any of the given categories.

Returns False if code is empty. Matching is case-insensitive.

For example:

>>> with open('script.sieve') as script:
>>>     try:
>>>         mgr.putscript(script, script.name)
>>>     except SieveOperationError as err:
>>>         if err.matches('QUOTA'):
>>>             print('over quota')

Print more informative messages:

>>> with open('script.sieve') as script:
>>>     try:
>>>         mgr.putscript(script, script.name)
>>>     except SieveOperationError as err:
>>>         if err.matches('QUOTA/MAXSCRIPTS'):
>>>             print('too many scripts')
>>>         elif err.matches('QUOTA/MAXSIZE'):
>>>             print(f'{script.name} is too large')
>>>         elif err.matches('QUOTA'):
>>>             print('over quota')
toerror() SieveError[source]

Convert a Response into an error.

code: tuple[Word, ...] = ()

Response code.

ManageSieve response codes are lists of categories, separated by slashes (“/”), where each category is the super-category of the next (e.g., “quota/maxsize”).

Some response codes carry data (e.g., TAG "SYNC-123").

See RFC 5804 (sec. 1.3) for a list of response codes.

Warning

Servers need not return response codes.

message: str | None = None

Human-readable message.

Warning

Servers need not return a message.

response: Atom

‘OK’, ‘NO’, or ‘BYE’.

Response

Meaning

‘OK’

Success

‘NO’

Failure

‘BYE’

Connection closed by server

class sievemgr.ResponseT

Type variable for Response.

alias of TypeVar(‘ResponseT’, bound=Response)

class sievemgr.SASLMechVar[source]

Bases: BaseVar, ListVarMixin

Convert SASL mechanism names to BaseAuth subclasses.

enum sievemgr.SASLPrep(value)[source]

Bases: IntEnum

Controls which strings are prepared for authentication.

See also

RFC 3454

Preparation of Internationalized Strings

RFC 4013

Stringprep Profile for User Names and Passwords

RFC 4422 (sec. 4)

SASL protocol requirements

Member Type:

int

Valid values are as follows:

NONE = <SASLPrep.NONE: 0>
USERNAMES = <SASLPrep.USERNAMES: 1>
PASSWORDS = <SASLPrep.PASSWORDS: 2>
ALL = <SASLPrep.ALL: 3>

The Enum and its members also have the following methods:

__new__(value)
class sievemgr.SRV[source]

Bases: object

DNS SRV record.

See also

RFC 2782

DNS SRV

__init__(priority: int, weight: int, host: str, port: int) None
host: str
port: int
priority: int
weight: int
class sievemgr.ScramSHA1Auth[source]

Bases: BaseScramAuth

SCRAM-SHA-1 authentication.

property digest: str

Digest name as used by hashlib and hmac.

name: ClassVar[str] = 'SCRAM-SHA-1'

Mechanism name.

order: int = -10

Mechanism precedence.

class sievemgr.ScramSHA1PlusAuth[source]

Bases: BaseScramPlusAuth, ScramSHA1Auth

SCRAM-SHA-1-PLUS authentication.

name: ClassVar[str] = 'SCRAM-SHA-1-PLUS'

Mechanism name.

order: int = -1000

Mechanism precedence.

class sievemgr.ScramSHA224Auth[source]

Bases: BaseScramAuth

SCRAM-SHA-224 authentication.

property digest: str

Digest name as used by hashlib and hmac.

name: ClassVar[str] = 'SCRAM-SHA-224'

Mechanism name.

order: int = -20

Mechanism precedence.

class sievemgr.ScramSHA224PlusAuth[source]

Bases: BaseScramPlusAuth, ScramSHA224Auth

SCRAM-SHA-224-PLUS authentication.

name: ClassVar[str] = 'SCRAM-SHA-224-PLUS'

Mechanism name.

order: int = -2000

Mechanism precedence.

class sievemgr.ScramSHA256Auth[source]

Bases: BaseScramAuth

SCRAM-SHA-256 authentication.

property digest: str

Digest name as used by hashlib and hmac.

name: ClassVar[str] = 'SCRAM-SHA-256'

Mechanism name.

order: int = -30

Mechanism precedence.

class sievemgr.ScramSHA256PlusAuth[source]

Bases: BaseScramPlusAuth, ScramSHA256Auth

SCRAM-SHA-256-PLUS authentication.

name: ClassVar[str] = 'SCRAM-SHA-256-PLUS'

Mechanism name.

order: int = -3000

Mechanism precedence.

class sievemgr.ScramSHA384Auth[source]

Bases: BaseScramAuth

SCRAM-SHA-384 authentication.

property digest: str

Digest name as used by hashlib and hmac.

name: ClassVar[str] = 'SCRAM-SHA-384'

Mechanism name.

order: int = -40

Mechanism precedence.

class sievemgr.ScramSHA384PlusAuth[source]

Bases: BaseScramPlusAuth, ScramSHA384Auth

SCRAM-SHA-384-PLUS authentication.

name: ClassVar[str] = 'SCRAM-SHA-384-PLUS'

Mechanism name.

order: int = -4000

Mechanism precedence.

class sievemgr.ScramSHA3_512Auth[source]

Bases: BaseScramAuth

SCRAM-SHA-512 authentication.

property digest

Digest name as used by hashlib and hmac.

name: ClassVar[str] = 'SCRAM-SHA3-512'

Mechanism name.

order: int = -60

Mechanism precedence.

class sievemgr.ScramSHA3_512PlusAuth[source]

Bases: BaseScramPlusAuth, ScramSHA3_512Auth

SCRAM-SHA-512-PLUS authentication.

name: ClassVar[str] = 'SCRAM-SHA3-512-PLUS'

Mechanism name.

order: int = -6000

Mechanism precedence.

class sievemgr.ScramSHA512Auth[source]

Bases: BaseScramAuth

SCRAM-SHA-512 authentication.

property digest

Digest name as used by hashlib and hmac.

name: ClassVar[str] = 'SCRAM-SHA-512'

Mechanism name.

order: int = -50

Mechanism precedence.

class sievemgr.ScramSHA512PlusAuth[source]

Bases: BaseScramPlusAuth, ScramSHA512Auth

SCRAM-SHA-512-PLUS authentication.

name: ClassVar[str] = 'SCRAM-SHA-512-PLUS'

Mechanism name.

order: int = -5000

Mechanism precedence.

enum sievemgr.ShellCmd(value)[source]

Bases: IntEnum

Shell actions that may overwrite or remove files.

Member Type:

int

Valid values are as follows:

NONE = <ShellCmd.NONE: 0>
CP = <ShellCmd.CP: 1>
GET = <ShellCmd.GET: 2>
MV = <ShellCmd.MV: 4>
PUT = <ShellCmd.PUT: 8>
RM = <ShellCmd.RM: 16>
ALL = <ShellCmd.ALL: 31>

The Enum and its members also have the following methods:

__new__(value)
class sievemgr.ShellPattern[source]

Bases: str

BaseShell pattern.

class sievemgr.SieveConfig[source]

Bases: BaseConfig

Configuration for the SieveManager command-line client.

__init__(*args, **kwargs)[source]

Create a new configuration.

Parameters:
  • args – Positional arguments used to initialise the back-end.

  • kwargs – Keyword arguments used as initial configuration values.

classmethod fromfiles(*fnames: str) SieveConfigT[source]

Create a new configuration from fnames.

Parameters:

fnames – Filenames (default: CONFIGFILES)

Raises:

FileNotFoundError – A given file could not be found.

getmanager(**variables) SieveManager[source]

Open a SieveManager connection with this configuration.

Parameters:

variables – Configuration variables.

Raises:
getshell(manager: SieveManager, **variables)[source]

Get a configured SieveShell that wraps manager.

loadaccount(host: str = 'localhost', login: str | None = None)[source]

Load the section for login on host.

loadfile(fname: str)[source]

Read configuration from fname.

Raises:
alias: UniqueVar

Alias for a host.

backups

How many backups to keep.

cadir

Custom CA directory.

cafile

Custom CA file.

cert

Client TLS certificate.

clobber

Overwrite files?

confirm

Which shell commands have to be confirmed?

getpassphrase

Command that prints the passphrase for the TLS key.

getpassword

Command that prints a password.

host

Host to connect to by default.

key

Client TLS key.

login

User to login as (authentication ID).

memory

How much memory to use for temporary data.

netrc: str | None = None

Filename of the .netrc file.

ocsp

Check whether server certificate was revoked?

owner

User whose scripts to manage (authorisation ID).

password

Password to login with.

port

Port to connect to by default.

saslmechs

How to authenticate.

saslprep

Which credentials to prepare.

timeout

Network timeout.

tls

Use TLS?

verbosity

Logging level.

x509strict

Be strict when verifying TLS certificates?

class sievemgr.SieveConn[source]

Bases: BaseACAPConn

Low-level connection to a ManageSieve server.

For example:

>>> conn = SieveConn('imap.foo.example')
>>> conn.authenticate('user', 'password')
>>> with open('script.sieve', 'br') as file:
>>>     conn.execute('putscript', file.name, file)
>>> conn.execute('logout')

Warning

SieveConn is not thread-safe.

See also

RFC 2244

Application Configuration Access Protocol

RFC 2782

DNS SRV

RFC 5804

ManageSieve

__init__(*args, **kwargs)[source]

Create a SieveConn object.

If args or kwargs are given, they are passed to open(). Otherwise, no connection is established.

For example:

>>> with SieveConn('imap.host.example') as conn:
>>>     conn.authenticate('user', 'password')
>>>     ...
>>> with SieveConn() as conn:
>>>     conn.open('imap.host.example')
>>>     conn.authenticate('user', 'password')
>>>     ...
Parameters:
  • args – Positional arguments for open().

  • kwargs – Keyword arguments for open().

Raises:
authenticate(login: str, *auth, owner: str = '', sasl: type[AbstractAuth] | Iterable[type[AbstractAuth]] = (), logauth: bool = False, **kwauth)[source]

Authenticate as login.

How the user is authenticated depends on the type of SASL_ mechanisms given in sasl (e.g., password-based or the “EXTERNAL” mechanism).

If no mechanisms are given, authentication is attempted with every supported non-obsolete password-based mechanism, starting with those with better security properties and progressing to those with worse security properties.

Unrecognised arguments are passed on to SASL mechanism constructors. Password-based mechanisms require a password:

>>> mgr.authenticate('user', 'password')

By contrast, the “EXTERNAL” mechanism takes no arguments:

>>> mgr.authenticate('user', sasl=ExternalAuth)

If an owner is given, the scripts of that owner are managed, instead of those owned by login. This requires elevated privileges.

Parameters:
  • login – User to login as (authentication ID).

  • owner – User whose scripts to manage (authorisation ID).

  • sasl – SASL mechanisms (default: BasePwdAuth.getmechs()).

  • logauth – Log authentication exchange?

  • auth – Positional arguments for SASL mechanism constructors.

  • kwauth – Keyword arguments for SASL mechanism constructors.

Raises:

Note

If an owner is given, but the selected authentication mechanism does not support proxy authentication, an error is logged to the console and authentication is attempted with the next mechanism.

close()[source]

Close the client side of the connection.

Warning

Call only when the server has closed the connection.

collect(check: bool = False) tuple[Response, list[Line]][source]

Collect the server’s response to the last command.

For example:

>>> conn.sendline(Atom('listscripts'))
>>> conn.collect()
(Response(response=Atom('OK'), code=(), message=None),
 [['foo.sieve', 'ACTIVE'], ['bar.sieve'], ['baz.sieve']])
Parameters:

check – Raise an error if the response is not “OK”?

Raises:
execute(command: str, *args: IO | Word) tuple[Response, list[Line]][source]

Execute command and return the server’s response.

For example:

>>> conn.execute('listscripts')
(Response(response=Atom('OK'), code=(), message=None),
 [['foo.sieve', 'ACTIVE'], ['bar.sieve'], ['baz.sieve']])
Raises:

Note

Referrals are followed automatically.

geturl() URL | None[source]

URL of the current connection.

For example:

>>> with SieveManager('imap.foo.example') as mgr:
>>>     mgr.authenticate('user', 'password')
>>>     mgr.geturl()
'sieve://user@imap.foo.example'

Note

Only changes to the connection state effected by open(), close(), shutdown(), authenticate(), unauthenticate(), and referrals are tracked.

isalive(check: bool = False) bool[source]

Check whether sock is alive.

Parameters:

check – Raise an error if sock has died.

Raises:

AppConnectionErrorsock has died. [5]

open(host: str, port: int = 4190, source: tuple[str, int] = ('', 0), timeout: float | None = socket.getdefaulttimeout(), tls: bool = True, ocsp: bool = True)[source]

Connect to host at port.

Parameters:
  • host – Server name or address.

  • port – Server port.

  • source – Source address and port.

  • timeout – Timeout in seconds.

  • tls – Secure the connection?

  • ocsp – Check whether the server certificate was revoked?

Raises:
shutdown()[source]

Shut the connection down.

Note

Use only when logging out would be unsafe.

capabilities: Capabilities | None = None

Server capabilities.

file: IO | BufferedRWPair | LogIOWrapper | None = None

File-like access to sock.

host: str | None = None

Remote address.

lock: allocate_lock = <unlocked _thread.lock object>

Operation lock.

logger: Logger = <Logger sievemgr (WARNING)>

Logger to use.

Messages are logged with the following priorities:

Priority

Used for

logging.ERROR

Non-fatal errors

logging.INFO

State changes

logging.DEBUG

Data sent to/received from the server

Suppress logging:

>>> from logging import getLogger
>>> getLogger('sievemgr').setLevel(logging.CRITICAL)

Use a custom logger:

>>> from logging import getLogger
>>> mgr.logger = getLogger('foo').addHandler(logging.NullHandler())

Print data send to/received from the server to standard error:

>>> from logging import getLogger
>>> getLogger('sievemgr').setLevel(logging.DEBUG)
>>> mgr.listscripts()
C: LISTSCRIPTS
S: "foo.sieve" ACTIVE
S: "bar.sieve"
S: "baz.sieve"
S: OK "Listscripts completed"
(Response(response=Atom('OK'), code=(), message=None),
 [('foo.sieve', True), ('bar.sieve', False), ('baz.sieve', False)])
login: str = ''

Login name (authentication ID).

ocsp: bool

Check whether the server certificate was revoked?

owner: str = ''

User whose scripts are managed (authorisation ID).

poll: poll | None = None

Polling object for sock.

port: int | None = None

Remote port.

sock: socket | None = None

Underlying socket.

sslcontext: SSLContext = <ssl.SSLContext object>

Settings for negotiating Transport Layer Security (TLS).

Disable workarounds for broken X.509 certificates:

>>> with SieveManager() as mgr:
>>>     mgr.sslcontext.verify_flags |= ssl.VERIFY_X509_STRICT
>>>     mgr.open('imap.foo.example')
>>>     ...

Load client certificate/key pair:

>>> with SieveManager() as mgr:
>>>     mgr.sslcontext.load_cert_chain(cert='cert.pem')
>>>     mgr.open('imap.foo.example')
>>>     ...

Use a custom certificate authority:

>>> with SieveManager() as mgr:
>>>     mgr.sslcontext.load_verify_locations(cafile='ca.pem')
>>>     mgr.open('imap.foo.example')
>>>     ...
property timeout: float | None

Connection timeout in seconds.

Set timeout to 500 ms:

>>> mgr.timeout = 0.5

Note

The timeout can only be set while a connection is open.

property tls: str | None

TLS version.

warning: str | None = None

Warning issued in response to the last “CHECKSCRIPT” or “PUTSCRIPT”.

For example:

>>> with open('script.sieve', 'br') as file:
>>>     mgr.execute('putscript', file, 'script.sieve')
(Response(response=Atom('OK'), code=('warnings,'),
 message='line 7: may need to be frobnicated'), [])
>>> mgr.warning
'line 7: may need to be frobnicated'

Note

Only set by collect(), execute(), checkscript(), and putscript().

See also

RFC 5804 (sec. 1.3)

ManageSieve “WARNINGS” response code.

class sievemgr.SieveManager[source]

Bases: SieveConn, AbstractContextManager

Connection to a ManageSieve server.

For example:

>>> with SieveManager('imap.foo.example') as mgr:
>>>     mgr.authenticate('user', 'password')
>>>     with open('sieve.script', 'br') as script:
>>>         mgr.putscript(script, 'sieve.script')
>>>     mgr.setactive('sieve.script')

Warning

SieveManager is not thread-safe.

__init__(*args, backup: int = 0, memory: int = 524_288, **kwargs)[source]

Create a SieveManager object.

If args or kwargs are given, they are passed to open(). Otherwise, no connection is established.

Parameters:
Raises:
backupscript(script: str, keep: int = 1)[source]

Make an Emacs-style backup of script.

keep = 0

Do nothing.

keep = 1

script is backed up as script~.

keep > 1

script is backed up as script.~n~. n starts with 1 and increments with each backup. Old backups are deleted if there are more than keep backups.

For example:

>>> mgr.listscripts()
[('script.sieve', True)]
>>> mgr.backupscript('script.sieve', keep=0)
>>> mgr.listscripts()
[('script.sieve', True)]
>>> mgr.listscripts()
[('script.sieve', True)]
>>> mgr.backupscript('script.sieve', keep=1)
>>> mgr.listscripts()
[('script.sieve', True), ('script.sieve~', False)]
>>> mgr.listscripts()
[('script.sieve', True)]
>>> mgr.backupscript('script.sieve', keep=2)
>>> mgr.listscripts()
[('script.sieve', True), ('script.sieve.~1~', False)]
>>> mgr.backupscript('script.sieve', keep=2)
>>> mgr.listscripts()
[('script.sieve', True),
 ('script.sieve.~1~', False),
 ('script.sieve.~2~', False)]
>>> mgr.backupscript('script.sieve', keep=2)
>>> mgr.listscripts()
[('script.sieve', True),
 ('script.sieve.~2~', False),
 ('script.sieve.~3~', False)]
Parameters:
  • script – Script name.

  • keep – How many backups to keep.

Raises:
checkscript(script: str | IO)[source]

Check whether script is valid.

Syntax errors trigger a SieveOperationError. Semantic errors are reported in warning.

For example:

>>> checkscript('foo')
Traceback (most recent call last):
    [...]
SieveOperationError: line 1: error: expected end of command ';'
error: parse failed.
>>> checkscript('# foo')
>>>
Parameters:

script – Script (not script name).

Raises:

Important

Sieve scripts must be encoded in UTF-8.

copyscript(source: str, target: str, backup: int | None = None)[source]

Download source and re-upload it as target.

Parameters:
  • source – Source name.

  • target – Target name.

  • backup – How many backups to keep (default: backup).

Raises:
deletescript(script: str)[source]

Delete script.

Raises:
editscripts(command: list[str], scripts: list[str], *args, catch: Callable[[Exception, str], bool] | None = None, check: bool = True, create: bool = True, **kwargs) CompletedProcess[source]

Download scripts, edit them with command, and re-upload them.

The scripts are appended to the command, which is then passed to subprocess.run(). Scripts that have been changed are then re-uploaded to the server. If the server has closed the connection in the meantime, the connection is re-established automatically.

If putscript() raises an error and catch has been given, then the error and the name of the offending script are passed to catch, which should return True if the command should be re-invoked for that script and False otherwise. Either way, the error will be suppressed.

For example:

>>> mgr.editscripts(['vi'], ['foo.sieve'])
>>> cp = mgr.editscripts(['cmp'], ['a.sieve', 'b.sieve'], check=False)
>>> if cp.returncode != 0:
>>>     print('a.sieve and b.sieve differ')
Parameters:
  • command – Command to run.

  • scripts – Scripts to edit.

  • catch – Error handler.

  • check – See subprocess.run().

  • create – Create scripts that do not exist?

  • args – Positional arguments for subprocess.run().

  • kwargs – Keywords arguments for subprocess.run().

Raises:
getactive() str | None[source]

Get the name of the active script.

Raises:
getscript(script: str) str[source]

Download script.

For example:

>>> with open('foo.sieve', 'w', encoding='utf8') as file:
>>>     file.write(mgr.getscript('foo.sieve'))
Parameters:

script – Script name.

Raises:
havespace(script: str, size: int)[source]

Check whether there is enough space for script.

Parameters:
  • script – Script name.

  • size – Script size in bytes.

Raises:
listscripts(cached: bool = False) list[tuple[str, bool]][source]

List scripts and whether they are the active script.

For example:

>>> mgr.listscripts()
[('foo.sieve', False), ('bar.sieve', True)]
>>> scripts = [script for script, _ in mgr.listscripts()]
Parameters:

cached – Return cached response? [7]

Returns:

A list of script name/status tuples.

Raises:
logout()[source]

Log out.

Note

logout() should be called to close the connection unless SieveManager is used as a context manager.

Warning

Logging out is unsafe after a ProtocolError. Use shutdown() instead.

noop(tag: str | None = None) str | None[source]

Request a no-op.

For example:

>>> mgr.noop('foo')
'foo'
Parameters:

tag – String for the server to echo back.

Returns:

Server echo.

Raises:
putscript(source: str | IO, target: str, backup: int | None = None)[source]

Upload source to the server as target.

The server should reject syntactically invalid scripts. It may issue a warning for semantically invalid scripts, but should accept them nonetheless. Updates are atomic.

For example:

>>> mgr.putscript('# empty', 'foo.sieve')
>>> with open('foo.sieve', 'br') as file:
>>>     mgr.putscript(file, 'foo.sieve')
Parameters:
  • source – Script (not script name).

  • target – Script name.

  • backup – How many backups to keep (default: backup).

Raises:

Important

Sieve scripts must be encoded in UTF-8.

renamescript(source: str, target: str, emulate: bool = True)[source]

Rename source to target.

Some servers do not the support the “RENAMESCRIPT” command. On such servers, renaming is emulated by downloading source, re-uploading it as target, marking target as the active script if source is the active script, and then deleting source.

For example:

>>> mgr.renamescript('foo.sieve', 'bar.sieve', emulate=False)
Parameters:
  • source – Script name.

  • target – Script name.

  • emulate – Emulate “RENAMESCRIPT” if the server does not support it?

Raises:
scriptexists(script: str, cached: bool = False) bool[source]

Check if script exists.

Parameters:
  • script – Script name.

  • cached – Return cached response? [7]

Raises:
setactive(script: str)[source]

Mark script as the active script.

Raises:
unauthenticate()[source]

Unauthenticate.

Raises:
unsetactive()[source]

Deactivate the active script.

Raises:
classmethod validname(script: str, check: bool = False) bool[source]

Check whether script is a valid script name.

Parameters:
  • script – Script name

  • check – Raise an error if script is not a valid script name?

Raises:

ValueErrorscript is not valid. [9]

backup: int = 0

How many backups to keep.

class sievemgr.SieveSASLAdapter[source]

Bases: AbstractSASLAdapter

Adapter to send SASL messages over a SieveConn.

__init__(connection: SieveConn)[source]

Initialise the adapter.

abort()[source]

Abort authentication.

Raises:

ProtocolError – Protocol violation.

begin(name: str, data: bytes | None = None)[source]

Begin authentication.

Parameters:
  • name – SASL mechanism name.

  • data – Optional client-first message.

Raises:
end()[source]

Conclude authentication.

Raises:
receive() bytes[source]

Receive and decode an SASL message.

Raises:
send(data: bytes)[source]

Encode and send an SASL message.

Raises:
conn: SieveConn | None = None

Underlying connection.

property sock: socket | SSLSocket

Underlying socket.

class sievemgr.SieveShell[source]

Bases: BaseShell

Shell around a SieveManager connection.

__init__(manager: SieveManager, clobber: bool = True, confirm: ShellCmd = ShellCmd.ALL)[source]

Initialise a SieveShell object.

Parameters:
  • manager – Connection to a ManageSieve server.

  • clobber – Overwrite files?

  • confirm – Shell commands that require confirmation.

complete_activate(*_) list[tuple[str, bool]]

Completer for activate.

complete_cat(*_) list[tuple[str, bool]]

Completer for cat.

static complete_cd(_: int, text: str) list[tuple[str, bool]]

Completer for cd.

static complete_check(_: int, text: str) list[tuple[str, bool]]

Completer for check.

complete_cmp(*_) list[tuple[str, bool]]

Completer for cmp.

complete_cp(*_) list[tuple[str, bool]]

Completer for cp.

complete_diff(*_) list[tuple[str, bool]]

Completer for diff.

static complete_dirs(_: int, text: str) list[tuple[str, bool]][source]

Complete local directory names.

complete_ed(*_) list[tuple[str, bool]]

Completer for ed.

static complete_files(_: int, text: str) list[tuple[str, bool]][source]

Complete local filenames.

complete_get(*_) list[tuple[str, bool]]

Completer for get.

complete_ls(*_) list[tuple[str, bool]]

Completer for ls.

complete_more(*_) list[tuple[str, bool]]

Completer for more.

complete_mv(*_) list[tuple[str, bool]]

Completer for mv.

static complete_put(_: int, text: str) list[tuple[str, bool]]

Completer for put.

complete_rm(*_) list[tuple[str, bool]]

Completer for rm.

complete_scripts(*_) list[tuple[str, bool]][source]

Complete script names.

complete_vi(*_) list[tuple[str, bool]]

Completer for vi.

static do_about()[source]

about - show information about SieveManager

do_activate(script: str)[source]

activate script - mark script as active

do_caps()[source]

caps - show server capabilities

do_cat(*scripts: str)[source]

cat [script …] - concatenate scripts on standard output

do_cd(localdir: str = HOME)[source]

cd [localdir] - change local directory

do_cert()[source]

cert - show the server’s TLS certificate.

do_check(localscript: str)[source]

check localscript - check whether localscript is valid

do_cmp(*args: str) int[source]

cmp [-s] script1 […] scriptN - compare scripts

do_cp(*args: str)[source]

cp [-f|-i] source target - re-upload source as target

do_deactivate()[source]

deactivate - deactivate the active script

do_diff(*args: str) int[source]

diff <options> script1 script2 - show how scripts differ

do_echo(*args: str)[source]

echo word […] - print words to standard output.

do_ed(*args: str)[source]

ed [-a] script […] - edit scripts with a line editor

do_get(*args: str)[source]

get [-a] [-f|-i] [-o file] [script …] - download script

do_ls(*args: str)[source]

ls [-1al] [script …] - list scripts

do_more(*args: str)[source]

more <options> script […] - display scripts page-by-page.

do_mv(*args: str)[source]

mv [-f|-i] source target - rename source to target

do_put(*args: str)[source]

put [-f|-i] [-a] [-o name] [localscript …] - upload scripts

do_python()[source]

python - enter Python read-evaluate-print loop

do_rm(*args: str)[source]

rm [-f|-i] [script …] - remove script

do_sh(*args: str)[source]

sh [command] [argument …] - run system command or system shell

do_su(user: str)[source]

su user - manage scripts of user.

do_vi(*args: str)[source]

vi [-a] script […] - edit scripts with a visual editor

do_xargs(command: str, *args: str)[source]

xargs cmd [arg …] - call cmd with arguments from standard input

editscripts(editor: list[str], *args: str)[source]

Edit scripts with the given editor.

enter() int[source]

Start reading commands from standard input.

Reading stops at the end-of-file marker or when a command raises StopIteration.

Side-effects

Entering the shell binds Tab to readline.complete().

execute(*args, **kwargs) int[source]

Execute command.

For example:

>>> shell.execute('ls', 'foo', 'bar')
0
Parameters:
  • command – Command name.

  • args – Arguments to the command.

Returns:

Return value.

Raises:

ShellUsageError – Command not found or arguments invalid.

getprompt(*args, **kwargs)[source]

Get a shell prompt.

clobber: bool

Overwrite files?

manager: SieveManager

Connection to a ManageSieve server.

reqconfirm: ShellCmd

Commands that require confirmation.

class sievemgr.T

Type variable.

alias of TypeVar(‘T’)

class sievemgr.TermIO[source]

Bases: TextIOWrapper

I/O for the controlling terminal.

__init__(*args, **kwargs)[source]

Open the controlling terminal.

class sievemgr.URL[source]

Bases: object

Sieve URL.

See also

RFC 5804 (sec. 3)

Sieve URL Scheme

__init__(hostname: str, scheme: str = 'sieve', username: str | None = None, password: str | None = None, port: int | None = None, owner: str | None = None, scriptname: str | None = None) None
classmethod fromstr(url: str) URLT[source]

Create a URL object from a URL string.

For example:

>>> URL.fromstr('sieve://user@imap.foo.example')
URL(hostname='imap.foo.example', scheme='sieve',
    username='user', password=None, port=None,
    owner=None, scriptname=None)
Raises:

ValueError – Not a valid Sieve URL.

hostname: str
owner: str | None = None
password: str | None = None
port: int | None = None
scheme: str = 'sieve'
scriptname: str | None = None
username: str | None = None
class sievemgr.URLT

Type variable for URL.

alias of TypeVar(‘URLT’, bound=URL)

class sievemgr.UniqueVar[source]

Bases: BaseVar

Variable the value of which must be unique.

values: ClassVar[set] = {}
sievemgr.askpass(prompt: str) str[source]

Prompt for a password on the controlling terminal.

sievemgr.backup(file: str, keep: int, getfiles: Callable[[], Iterable[str]], copy: Callable[[str, str], Any], remove: Callable[[str], Any])[source]

Make an Emacs-style backup of file.

keep = 1

file is backed up as file~.

keep > 1

file is backed up as file.~n~, where n starts with 1 and increments with each backup.

Parameters:
  • file – File to back up.

  • keep – How many copies to keep.

  • copy – Function that copies the file.

  • getfiles – Function that returns a list of files.

  • remove – Function that removes a file.

Raises:

ValueErrorkeep is < 0.

sievemgr.bell()[source]

Print a BEL to the controlling terminal, if there is one.

sievemgr.certrevoked(cert, logger: Logger = logging.getLogger(__name__)) bool[source]

Check if cert has been revoked.

Raises:

See also

RFC 5019

Lightweight OCSP Profile

RFC 6960

Online Certificate Status Protocol (OCSP)

sievemgr.error(*args, status: int = 1, **kwargs) NoReturn[source]

Log an err and exit with status.

Parameters:
sievemgr.escapectrl(chars: str) str[source]

Escape control characters.

sievemgr.getcertauthinfo(cert) tuple[list[str], list[str]][source]

Get information about the authority that issued cert.

Returns:

CA issuer URLs and OCSP responder base URLs.

sievemgr.getfilesize(file: IO) int[source]

Get the size of file-like object relative to the current position.

sievemgr.httpget(url: str) bytes[source]

Download a file from url using HTTP.

Raises:
  • HTTPUsageErrorurl is not an HTTP URL.

  • HTTPOperationError – “GET” failed.

sievemgr.isdnsname(name: str) bool[source]

Check whether name is a valid DNS name.

See also

RFC 1035 (sec. 2.3.1)

Domain names - Preferred name syntax

RFC 2181 (sec. 11)

Clarifications to the DNS Specification - Name syntax

sievemgr.ishostname(name: str) bool[source]

Check whether name is a valid hostname.

See also

RFC 921

Domain Name System Implementation Schedule

RFC 952

Internet host table specification

RFC 1123 (sec. 2.1)

Host Names and Numbers

sievemgr.isinetaddr(addr: str) bool[source]

Check whether addr is an internet address.

sievemgr.main(*args, **kwargs) NoReturn[source]

sievemgr - manage remote Sieve scripts

Usage: sievemgr [server] [command] [argument …]

sievemgr -e expression […] [server] sievemgr -s file [server]

Options:
-C

Do not overwrite existing files.

-c file

Read configuration from file.

-d

Enable debugging mode.

-e expression

Execute expression on the server.

-f

Overwrite and remove files without confirmation.

-i

Confirm removing or overwriting files.

-o key=value Set configuration key to value. -q Be quieter. -s file Execute expressions read from file. -v Be more verbose.

-e, -o, -q, and -v can be given multiple times. See sievemgr(1) for the complete list.

Report bugs to: <https://github.com/odkr/sievemgr/issues> Home page: <https://odkr.codeberg.page/sievemgr>

sievemgr.nwise(iterable: Iterable, n: Any) Iterator[tuple][source]

Iterate over n-tuples.

sievemgr.randomise(elems: Sequence[T], weights: Iterable[int]) list[T][source]

Randomise the order of elems.

sievemgr.readdir(dirname: str, predicate: Callable[[str], bool] | None = None) Iterator[str][source]

Get every filename in dirname that matches predicate.

sievemgr.readnetrc(fname: str | None) dict[str, tuple[str, str, str]][source]

Read a .netrc file.

Parameters:

fname – Filename (default: ~/.netrc)

Returns:

Mapping from hosts to login-account-password 3-tuples.

Raises:
sievemgr.readoutput(*command: str, encoding: str = ENCODING, logger: Logger = logging.getLogger(__name__)) str[source]

Decode and return the output of command.

Returns:

Decoded output or, if command exited with a non-zero status, the empty string.

Raises:

subprocess.CalledProcessErrorcommand exited with a status >= 127.

sievemgr.resolvesrv(host: str) Iterator[SRV][source]

Resolve a DNS SRV record.

Parameters:

host – Hostname (e.g., _sieve._tcp.imap.foo.example)

Returns:

An iterator over SRV records sorted by their priority and randomised according to their weight.

Raises:
  • DNSDataErrorhost is not a valid DNS name.

  • DNSOperationError – Lookup error.

  • DNSSoftwareErrordnspython is not available.

Note

Requires dnspython.

sievemgr.showhelp(func: Callable) NoReturn[source]

Print the docstring of func and exit.

sievemgr.showversion() NoReturn[source]

Print ABOUT and exit.

sievemgr.yamlescape(data: str)[source]

Strip and quote data for use as YAML scalar if needed.

sievemgr.ABOUT: Final[str] = 'SieveManager 0.7.4.7\nCopyright 2023 and 2024 Odin Kroeger'

About message.

sievemgr.AuthMech

Alias for subclasses of AbstractAuth.

alias of type[AbstractAuth]

sievemgr.CONFIGFILES: Final[tuple[str, ...]] = ('/etc/sieve/config', '/etc/sieve.cf', '/Users/odin/.config/sieve/config', '/Users/odin/.sieve/config', '/Users/odin/.sieve.cf')

Default configuration files.

sievemgr.DEBUG: bool = False

Print stack traces even for expected error types?

sievemgr.EDITOR: Final[list[str]] = ['vim']

EDITOR or ed if EDITOR is unset.

sievemgr.ENCODING: Final[str] = 'UTF-8'

Encoding.

sievemgr.HOME: Final[str] = '/Users/odin'

Home directory.

sievemgr.Line

List of Word-s.

alias of list[Word]

sievemgr.PAGER: Final[list[str]] = ['less']

PAGER or more if PAGER is unset.

sievemgr.ShellWord

Alias for ShellPattern and str.

alias of ShellPattern | str

sievemgr.SignalHandler

Alias for signal handlers.

alias of Callable[[int, FrameType | None], Any] | int | None

sievemgr.SignalHandlingFunc

Alias for signal handling functions.

alias of Callable[[int, FrameType | None], Any]

sievemgr.VISUAL: Final[list[str]] = ['vim']

VISUAL or vi if VISUAL is unset.

sievemgr.Word

Alias for Atom, None, int, str, and Line.

alias of Atom | None | int | str | list[Word]

sievemgr.XDG_CONFIG_HOME: Final[str] = '/Users/odin/.config'

X Desktop group base configuration directory.