This file is indexed.

/usr/lib/python3/dist-packages/aiosmtpd/main.py is in python3-aiosmtpd 1.1-5.

This file is owned by root:root, with mode 0o644.

The actual contents of the file can be viewed below.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
import os
import sys
import signal
import asyncio
import logging

from aiosmtpd.smtp import DATA_SIZE_DEFAULT, SMTP, __version__
from argparse import ArgumentParser
from contextlib import suppress
from functools import partial
from importlib import import_module
from public import public

try:
    import pwd
except ImportError:                                         # pragma: nocover
    pwd = None


DEFAULT_HOST = 'localhost'
DEFAULT_PORT = 8025

# Make the program name a little nicer, especially when `python3 -m aiosmtpd`
# is used.
PROGRAM = 'smtpd' if '__main__.py' in sys.argv[0] else sys.argv[0]


def parseargs(args=None):
    parser = ArgumentParser(
        prog=PROGRAM,
        description='An RFC 5321 SMTP server with extensions.')
    parser.add_argument(
        '-v', '--version', action='version',
        version='%(prog)s {}'.format(__version__))
    parser.add_argument(
        '-n', '--nosetuid',
        dest='setuid', default=True, action='store_false',
        help="""This program generally tries to setuid `nobody`, unless this
                flag is set.  The setuid call will fail if this program is not
                run as root (in which case, use this flag).""")
    parser.add_argument(
        '-c', '--class',
        dest='classpath',
        default='aiosmtpd.handlers.Debugging',
        help="""Use the given class, as a Python dotted import path, as the
                handler class for SMTP events.  This class can process
                received messages and do other actions during the SMTP
                dialog.  Uses a debugging handler by default.""")
    parser.add_argument(
        '-s', '--size',
        type=int,
        help="""Restrict the total size of the incoming message to
                SIZE number of bytes via the RFC 1870 SIZE extension.
                Defaults to {} bytes.""".format(DATA_SIZE_DEFAULT))
    parser.add_argument(
        '-u', '--smtputf8',
        default=False, action='store_true',
        help="""Enable the SMTPUTF8 extension and behave as an RFC 6531
                SMTP proxy.""")
    parser.add_argument(
        '-d', '--debug',
        default=0, action='count',
        help="""Increase debugging output.""")
    parser.add_argument(
        '-l', '--listen', metavar='HOST:PORT',
        nargs='?', default=None,
        help="""Optional host and port to listen on.  If the PORT part is not
                given, then port {port} is used.  If only :PORT is given,
                then {host} is used for the hostname.  If neither are given,
                {host}:{port} is used.""".format(
                    host=DEFAULT_HOST, port=DEFAULT_PORT))
    parser.add_argument(
        'classargs', metavar='CLASSARGS',
        nargs='*', default=(),
        help="""Additional arguments passed to the handler CLASS.""")
    args = parser.parse_args(args)
    # Find the handler class.
    path, dot, name = args.classpath.rpartition('.')
    module = import_module(path)
    handler_class = getattr(module, name)
    if hasattr(handler_class, 'from_cli'):
        args.handler = handler_class.from_cli(parser, *args.classargs)
    else:
        if len(args.classargs) > 0:
            parser.error('Handler class {} takes no arguments'.format(path))
        args.handler = handler_class()
    # Parse the host:port argument.
    if args.listen is None:
        args.host = DEFAULT_HOST
        args.port = DEFAULT_PORT
    else:
        host, colon, port = args.listen.rpartition(':')
        if len(colon) == 0:
            args.host = port
            args.port = DEFAULT_PORT
        else:
            args.host = DEFAULT_HOST if len(host) == 0 else host
            try:
                args.port = int(DEFAULT_PORT if len(port) == 0 else port)
            except ValueError:
                parser.error('Invalid port number: {}'.format(port))
    return parser, args


@public
def main(args=None):
    parser, args = parseargs(args=args)

    if args.setuid:                               # pragma: nomswin
        if pwd is None:
            print('Cannot import module "pwd"; try running with -n option.',
                  file=sys.stderr)
            sys.exit(1)
        nobody = pwd.getpwnam('nobody').pw_uid
        try:
            os.setuid(nobody)
        except PermissionError:
            print('Cannot setuid "nobody"; try running with -n option.',
                  file=sys.stderr)
            sys.exit(1)

    factory = partial(
        SMTP, args.handler,
        data_size_limit=args.size, enable_SMTPUTF8=args.smtputf8)

    logging.basicConfig(level=logging.ERROR)
    log = logging.getLogger('mail.log')
    loop = asyncio.get_event_loop()

    if args.debug > 0:
        log.setLevel(logging.INFO)
    if args.debug > 1:
        log.setLevel(logging.DEBUG)
    if args.debug > 2:
        loop.set_debug(enabled=True)

    log.info('Server listening on %s:%s', args.host, args.port)

    server = loop.run_until_complete(
        loop.create_server(factory, host=args.host, port=args.port))
    # Signal handlers are only supported on *nix, so just ignore the failure
    # to set this on Windows.
    with suppress(NotImplementedError):
        loop.add_signal_handler(signal.SIGINT, loop.stop)

    log.info('Starting asyncio loop')
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass
    server.close()
    log.info('Completed asyncio loop')
    loop.run_until_complete(server.wait_closed())
    loop.close()


if __name__ == '__main__':                          # pragma: nocover
    main()