Compare commits
1 Commits
Author | SHA1 | Date | |
---|---|---|---|
2da8f13ebd |
@ -1,14 +1,14 @@
|
||||
import json
|
||||
import re
|
||||
import logging
|
||||
from typing import Callable, Union, Optional
|
||||
from typing import Callable, Union, Optional, List, Tuple
|
||||
from jata import Jata, MutableDefault
|
||||
|
||||
|
||||
class Match(Jata):
|
||||
name: str
|
||||
addrs: list[str] = MutableDefault(lambda: []) # type: ignore
|
||||
addr_rexs: list[str] = MutableDefault(lambda: []) # type: ignore
|
||||
addrs: List[str] = MutableDefault(lambda: []) # type: ignore
|
||||
addr_rexs: List[str] = MutableDefault(lambda: []) # type: ignore
|
||||
|
||||
|
||||
DEFAULT_MATCH_ALL = "default_match_all"
|
||||
@ -23,7 +23,7 @@ class Rule(Jata):
|
||||
|
||||
class Mbox(Jata):
|
||||
name: str
|
||||
rules: list[Rule]
|
||||
rules: List[Rule]
|
||||
|
||||
|
||||
DEFAULT_NULL_MBOX = "default_null_mbox"
|
||||
@ -77,18 +77,18 @@ class Config(Jata):
|
||||
logging: Optional[LogCfg] = None
|
||||
|
||||
mails_path: str
|
||||
matches: list[Match]
|
||||
boxes: list[Mbox]
|
||||
users: list[User]
|
||||
matches: List[Match]
|
||||
boxes: List[Mbox]
|
||||
users: List[User]
|
||||
|
||||
servers: list[ServerCfg]
|
||||
servers: List[ServerCfg]
|
||||
|
||||
|
||||
CheckerFn = Callable[[str], bool]
|
||||
Checker = tuple[str, CheckerFn, bool]
|
||||
Checker = Tuple[str, CheckerFn, bool]
|
||||
|
||||
|
||||
def parse_checkers(cfg: Config) -> list[Checker]:
|
||||
def parse_checkers(cfg: Config) -> List[Checker]:
|
||||
def make_match_fn(m: Match):
|
||||
if m.addrs and m.addr_rexs:
|
||||
raise Exception("Both addrs and addr_rexs is set")
|
||||
@ -118,7 +118,7 @@ def parse_checkers(cfg: Config) -> list[Checker]:
|
||||
]
|
||||
|
||||
|
||||
def get_mboxes(addr: str, checks: list[Checker]) -> list[str]:
|
||||
def get_mboxes(addr: str, checks: List[Checker]) -> List[str]:
|
||||
def inner():
|
||||
for mbox, match_fn, stop_check in checks:
|
||||
if match_fn(addr):
|
||||
@ -130,7 +130,7 @@ def get_mboxes(addr: str, checks: list[Checker]) -> list[str]:
|
||||
return list(inner())
|
||||
|
||||
|
||||
def gen_addr_to_mboxes(cfg: Config) -> Callable[[str], list[str]]:
|
||||
def gen_addr_to_mboxes(cfg: Config) -> Callable[[str], List[str]]:
|
||||
checks = parse_checkers(cfg)
|
||||
logging.info(f"Parsed checkers from config, {len(checks)=}")
|
||||
return lambda addr: get_mboxes(addr, checks)
|
||||
|
@ -13,7 +13,7 @@ from .pwhash import parse_hash, check_pass, PWInfo
|
||||
from asyncio import StreamReader, StreamWriter
|
||||
import random
|
||||
|
||||
from typing import Optional
|
||||
from typing import Optional, List, Tuple, Dict
|
||||
|
||||
from .poputils import (
|
||||
InvalidCommand,
|
||||
@ -46,7 +46,7 @@ class State:
|
||||
|
||||
|
||||
class SharedState:
|
||||
def __init__(self, mails_path: Path, users: dict[str, tuple[PWInfo, str]]):
|
||||
def __init__(self, mails_path: Path, users: dict[str, Tuple[PWInfo, str]]):
|
||||
self.mails_path = mails_path
|
||||
self.users = users
|
||||
self.loggedin_users: set[str] = set()
|
||||
@ -237,7 +237,7 @@ def trans_command_noop(_, __) -> None:
|
||||
write(ok("Hmm"))
|
||||
|
||||
|
||||
async def process_transactions(mails_list: list[MailEntry]) -> set[str]:
|
||||
async def process_transactions(mails_list: List[MailEntry]) -> set[str]:
|
||||
mails = MailList(mails_list)
|
||||
|
||||
def reset(_, __):
|
||||
@ -328,7 +328,7 @@ async def start_session() -> None:
|
||||
scfg().loggedin_users.remove(state().username)
|
||||
|
||||
|
||||
def parse_users(users: list[User]) -> dict[str, tuple[PWInfo, str]]:
|
||||
def parse_users(users: List[User]) -> Dict[str, Tuple[PWInfo, str]]:
|
||||
def inner():
|
||||
for user in users:
|
||||
user = User(user)
|
||||
@ -338,7 +338,7 @@ def parse_users(users: list[User]) -> dict[str, tuple[PWInfo, str]]:
|
||||
return dict(inner())
|
||||
|
||||
|
||||
def make_pop_server_callback(mails_path: Path, users: list[User], timeout_seconds: int):
|
||||
def make_pop_server_callback(mails_path: Path, users: List[User], timeout_seconds: int):
|
||||
scfg = SharedState(mails_path=mails_path, users=parse_users(users))
|
||||
|
||||
async def session_cb(reader: StreamReader, writer: StreamWriter):
|
||||
@ -362,7 +362,7 @@ async def create_pop_server(
|
||||
host: str,
|
||||
port: int,
|
||||
mails_path: Path,
|
||||
users: list[User],
|
||||
users: List[User],
|
||||
ssl_context: Optional[ssl.SSLContext] = None,
|
||||
timeout_seconds: int = 60,
|
||||
) -> asyncio.Server:
|
||||
|
@ -2,6 +2,7 @@ import os
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum, auto
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
|
||||
class ClientError(Exception):
|
||||
@ -112,13 +113,13 @@ def files_in_path(path):
|
||||
return []
|
||||
|
||||
|
||||
def get_mails_list(dirpath: Path) -> list[MailEntry]:
|
||||
def get_mails_list(dirpath: Path) -> List[MailEntry]:
|
||||
files = files_in_path(dirpath)
|
||||
entries = [MailEntry(filename, path) for filename, path in files]
|
||||
return entries
|
||||
|
||||
|
||||
def set_nid(entries: list[MailEntry]):
|
||||
def set_nid(entries: List[MailEntry]):
|
||||
entries.sort(reverse=True, key=lambda e: (e.c_time, e.uid))
|
||||
for i, entry in enumerate(entries, start=1):
|
||||
entry.nid = i
|
||||
@ -130,7 +131,7 @@ def get_mail(entry: MailEntry) -> bytes:
|
||||
|
||||
|
||||
class MailList:
|
||||
def __init__(self, entries: list[MailEntry]):
|
||||
def __init__(self, entries: List[MailEntry]):
|
||||
self.entries = entries
|
||||
set_nid(self.entries)
|
||||
self.mails_map = {str(e.nid): e for e in entries}
|
||||
|
@ -57,7 +57,7 @@ async def a_main(cfg: config.Config) -> None:
|
||||
return host
|
||||
|
||||
mbox_finder = config.gen_addr_to_mboxes(cfg)
|
||||
servers: list[asyncio.Server] = []
|
||||
servers: List[asyncio.Server] = []
|
||||
|
||||
if not cfg.servers:
|
||||
logging.warning("Nothing to do!")
|
||||
|
@ -7,7 +7,7 @@ import uuid
|
||||
import shutil
|
||||
from functools import partial
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
from typing import Callable, Optional, List
|
||||
from . import config
|
||||
from email.message import Message
|
||||
import email.policy
|
||||
@ -25,7 +25,7 @@ logger = logging.getLogger("smtp")
|
||||
|
||||
|
||||
class MyHandler(AsyncMessage):
|
||||
def __init__(self, mails_path: Path, mbox_finder: Callable[[str], list[str]]):
|
||||
def __init__(self, mails_path: Path, mbox_finder: Callable[[str], List[str]]):
|
||||
super().__init__()
|
||||
self.mails_path = mails_path
|
||||
self.mbox_finder = mbox_finder
|
||||
@ -63,7 +63,7 @@ class MyHandler(AsyncMessage):
|
||||
|
||||
|
||||
def protocol_factory_starttls(
|
||||
mails_path: Path, mbox_finder: Callable[[str], list[str]], context: ssl.SSLContext
|
||||
mails_path: Path, mbox_finder: Callable[[str], List[str]], context: ssl.SSLContext
|
||||
):
|
||||
logger.info("Got smtp client cb starttls")
|
||||
try:
|
||||
@ -80,7 +80,7 @@ def protocol_factory_starttls(
|
||||
return smtp
|
||||
|
||||
|
||||
def protocol_factory(mails_path: Path, mbox_finder: Callable[[str], list[str]]):
|
||||
def protocol_factory(mails_path: Path, mbox_finder: Callable[[str], List[str]]):
|
||||
logger.info("Got smtp client cb")
|
||||
try:
|
||||
handler = MyHandler(mails_path, mbox_finder)
|
||||
@ -95,9 +95,9 @@ async def create_smtp_server_starttls(
|
||||
host: str,
|
||||
port: int,
|
||||
mails_path: Path,
|
||||
mbox_finder: Callable[[str], list[str]],
|
||||
mbox_finder: Callable[[str], List[str]],
|
||||
ssl_context: ssl.SSLContext,
|
||||
) -> asyncio.Server:
|
||||
):
|
||||
logging.info(
|
||||
f"Starting SMTP STARTTLS server {host=}, {port=}, {mails_path=!s}, {ssl_context != None=}"
|
||||
)
|
||||
@ -114,9 +114,9 @@ async def create_smtp_server(
|
||||
host: str,
|
||||
port: int,
|
||||
mails_path: Path,
|
||||
mbox_finder: Callable[[str], list[str]],
|
||||
mbox_finder: Callable[[str], List[str]],
|
||||
ssl_context: Optional[ssl.SSLContext] = None,
|
||||
) -> asyncio.Server:
|
||||
):
|
||||
logging.info(
|
||||
f"Starting SMTP server {host=}, {port=}, {mails_path=!s}, {ssl_context != None=}"
|
||||
)
|
||||
|
Loading…
Reference in New Issue
Block a user