path: root/main.py

from email.message import EmailMessage
from email.headerregistry import Address
from urllib.parse import urlparse, urlencode, parse_qs
from http.cookies import BaseCookie
import http.cookies
import password
from password import Passwords
from uuid import uuid4
import os
from typing import (
    Optional,
    cast,
)
from mu import mu_search, get_mail
from html_render import HTML, render_document

from http.server import HTTPServer, BaseHTTPRequestHandler


def mailto(addr: str) -> HTML:
    return ('a', {'href': f'mailto:{addr}'}, addr)


def format_email(addr: Address) -> list[HTML]:
    mail_addr = f'{addr.username}@{addr.domain}'
    return [addr.display_name, ' <', mailto(mail_addr), '>']


def header_format(key: str, value) -> HTML:
    if key in ['to', 'cc', 'bcc']:
        return ('ul', *[('li', *format_email(addr))
                        for addr in value.addresses])
    elif key == 'from':
        return format_email(value.addresses[0])
    elif key == 'in-reply-to':
        # type(value) == email.headerregistry._UnstructuredHeader
        id = str(value).strip("<>")
        return ['<', ('a', {'href': '?' + urlencode({'id': id})}, id), '>']
    else:
        return value


style: HTML = lambda: """
    nav {
        display: block;
        width: 100%;
        height: 4em;
        color: white;
        background-color: darkgrey;
    }

    dl {
        display: grid;
        grid-template-columns: 10ch auto;
    }
    dt {
        font-weight: bold;
    }
    dd {
        font-family: mono;
        font-size: 80%;
    }
    dd > * {
        margin: 0;
    }

    """


def attachement_tree(mail: EmailMessage) -> HTML:
    ct = mail.get_content_type()
    fn = mail.get_filename()

    children = []
    for child in mail.iter_parts():
        children.append(attachement_tree(cast(EmailMessage, child)))

    content: HTML
    if children:
        content = ('ul', *children)
    else:
        content = []

    if fn:
        body = f'{ct} {fn}'
    else:
        body = str(ct)
    return ('li', body, content)

# --------------------------------------------------


def login_page(returnto: Optional[str] = None) -> HTML:
    return ('form', {'action': '/login', 'method': 'POST'},
            ('input', {'name': 'username', 'placeholder': 'Username'}),
            ('input', {'type': 'password',
                       'placeholder': 'Password',
                       'name': 'password'}),
            ('input', {'type': 'hidden',
                       'name': 'returnto',
                       'value': returnto})
            if returnto else [],
            ('input', {'type': 'submit'}),
            )


def user_info(username: str) -> HTML:
    return [('span', username),
            ('form', {'action': '/logout', 'method': 'POST'},
                ('input', {'type': 'submit', 'value': 'Logga ut'}))]


def login_prompt() -> HTML:
    return ('a', {'href': '/login'}, 'Logga in')


def page_base(title: Optional[str] = None,
              body: HTML = [],
              username: Optional[str] = None) -> HTML:
    return ('html',
            ('head',
             ('meta', {'charset': 'utf-8'}),
             ('title', title),
             ('style', style),
             ),
            ('body',
             ('nav',
              user_info(username) if username else login_prompt()
              ),
             body))


def response_for(id: str, username: Optional[str] = None) -> str:

    mail = cast(EmailMessage, get_mail(id))

    headers = {}
    for (key, value) in mail.items():
        headers[key.lower()] = value

    head = []
    for h in ['date', 'from', 'to', 'cc', 'bcc', 'subject', 'x-original-to',
              'in-reply-to']:
        if x := headers.get(h.lower()):
            head += [('dt', h.title()),
                     ('dd', header_format(h.lower(), x))]

    body_part = mail.get_body(preferencelist=('html', 'plain'))
    if not body_part:
        raise ValueError("No suitable body in email")
    ct = body_part.get_content_type()
    body: HTML
    if ct == 'text/html':
        body = lambda: cast(EmailMessage, body_part).get_content()
    else:
        body = ('pre', cast(EmailMessage, body_part).get_content())

    if t := headers.get('subject'):
        title = f'Mail — {t}'
    else:
        title = 'Mail'

    main_body = [('dl', *head),
                 ('hr',),
                 ('main', body),
                 ('hr',),
                 ('ul', attachement_tree(mail)),
                 ]
    html_str = render_document(page_base(title=title,
                                         body=main_body,
                                         username=username))

    return html_str


def search_field(q: str) -> HTML:
    return ('form', {'action': '/search', 'method': 'GET'},
            ('label', {'for': 'search'},
             'Mu Search Query'),
            ('textarea', {'id': 'search', 'name': 'q'},
             q),
            ('input', {'type': 'Submit'}))


def search_result(q, by, reverse):

    keys = ['From', 'To', 'Subject', 'Date', 'Size', 'Maildir', 'Msgid']

    rows = mu_search(q, by, reverse)
    body = []
    for row in rows:
        rowdata = ['tr']
        for key in keys:
            rowdata.append(row[key])
        body.append(rowdata)

    return ('table',
            ('thead',
             ('tr',
              [('th', m) for m in keys])),
            ('tbody',
             body
             ))


def search_page(q, by, username=None):
    main_body = [search_field(q)]

    if q:
        main_body.append(search_result(q, by, False))

    return render_document(page_base(title='Serach',
                                     body=main_body,
                                     username=username))


def index_page(username):
    ids = [
        'CAEzixGsw-4zJ8_ejK_vDgmcQ9s-MbBc-ho+HL4arV4a+ghOOPg@mail.gmail.com',
        'CA+pcBt-gLb0GtbFOjJ5_7Q_WXtqApVPQ9w-3O7GH=VqCEQat6g@mail.gmail.com',
    ]

    body = [('h1', "Sample ID's"),
            ('ul',
             [('li', ('a', {'href': '?' + urlencode({'id': id})}, id))
              for id in ids]
             ),
            ]

    return render_document(page_base(title='Mail index',
                                     body=body,
                                     username=username
                                     ))


valid_session_cookies: dict[str, str] = {}


def validate_session_cookie(cookie: http.cookies.Morsel) -> Optional[str]:
    return valid_session_cookies.get(cookie.value)


def remove_session_cookie(cookie: http.cookies.Morsel) -> http.cookies.Morsel:
    if valid_session_cookies.get(cookie.value):
        del valid_session_cookies[cookie.value]
    cookie.set(cookie.key, '', '')
    # TODO how to expire cookie
    # cookie.expires = 0
    return cookie


passwords: Passwords = password.Passwords(cast(os.PathLike, 'passwords.json'))


def new_session_cookie(username: str) -> http.cookies.Morsel:
    global valid_session_cookies
    m: http.cookies.Morsel = http.cookies.Morsel()
    unique = str(uuid4())
    valid_session_cookies[unique] = username
    m.set('session', unique, unique)
    return m


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        url = urlparse(self.path)
        query = parse_qs(url.query)
        # print(type(self.headers))

        cookies = BaseCookie(self.headers.get('Cookie'))
        logged_in = None
        if c := cookies.get('session'):
            logged_in = validate_session_cookie(c)

        if url.path == '/':
            if not logged_in:
                self.send_response(307)
                q = urlencode({'returnto': self.path})
                self.send_header('location', '/login?' + q)
                self.end_headers()
            else:
                if id := query.get('id'):
                    print("id =", id)
                    response = response_for(''.join(id).replace(' ', '+'),
                                            logged_in)
                    self.send_response(200)
                else:
                    response = index_page(logged_in)
                    self.send_response(200)

                response = response.encode('UTF-8')
                self.send_header('Content-Type', 'text/html; charset=UTF-8')
                self.send_header('Content-Length', len(response))
                self.end_headers()
                self.wfile.write(response)

        elif url.path == '/search':
            if not logged_in:
                self.send_response(307)
                q = urlencode({'returnto': self.path})
                self.send_header('location', '/login?' + q)
                self.end_headers()
            else:
                response = search_page(query.get('q'),
                                       query.get('by'),
                                       logged_in)
                self.send_response(200)
                response = response.encode('UTF-8')
                self.send_header('Content-Type', 'text/html; charset=UTF-8')
                self.send_header('Content-Length', len(response))
                self.end_headers()
                self.wfile.write(response)

        elif url.path == '/login':
            if not logged_in:
                body = login_page(''.join(query.get('returnto')))
                self.send_response(200)
                content = render_document(page_base(title='Login', body=body))
                content = content.encode('UTF-8')
                self.send_header('Content-Type', 'text/html; charset=UTF-8')
                self.send_header('Content-Length', len(content))
                self.end_headers()
                self.wfile.write(content)
            else:
                # TODO do something sensible here
                pass

    def do_POST(self):
        url = urlparse(self.path)
        # query = parse_qs(url.query)
        cookies = BaseCookie(self.headers.get('Cookie'))
        logged_in = None
        if c := cookies.get('session'):
            logged_in = validate_session_cookie(c)
        print(cookies)
        print(valid_session_cookies)

        if url.path == '/login':
            # cl = content_length = self.headers.get('content-length')
            cl = self.headers.get('content-length')
            data = parse_qs(self.rfile.read(int(cl)))
            username = b''.join(data[b'username']).decode('UTF-8')
            password = b''.join(data[b'password']).decode('UTF-8')
            if passwords.validate(username, password):
                cookie = new_session_cookie(username)
                self.send_response(302)
                self.send_header('set-cookie', cookie.OutputString())
                if ret := data.get(b'returnto'):
                    returnto = b''.join(ret).decode('UTF-8')
                    self.send_header('location', returnto)
                else:
                    self.send_header('location', '/')
            else:
                self.send_response(302)
                self.send_header('location', '/')

            self.end_headers()

        if url.path == '/logout':
            if not logged_in:
                self.send_response(302)
                self.send_header('Location', '/')
                self.end_headers()
                return
            cookie = remove_session_cookie(cookies.get('session'))
            self.send_response(302)
            self.send_header('set-cookie', cookie)
            # TODO use the referer header?
            self.send_header('Location', '/')
            self.end_headers()


if __name__ == '__main__':
    server = HTTPServer(('0', 8090), Handler)

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass

    server.server_close()