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
|
# SPDX-License-Identifier: GPL-2.0+
import asyncio
import itertools
import logging
import os
from typing import TYPE_CHECKING, Any, Dict, List
import pyinotify
if TYPE_CHECKING:
from . import messages, mailbox, oauth
logger: logging.Logger
class Config(object):
"""Program configuration and general global state"""
message_db_dir = "~/mail/.cms/"
domains: Dict[str, Any] = {}
trace_file: Any = None
web_app: "oauth.WebServer"
logger: logging.Logger
loop: asyncio.AbstractEventLoop
watch_manager: pyinotify.WatchManager
msgdb: "messages.MessageDB"
cloud_mboxes: "List[mailbox.Mailbox]"
local_mboxes: "List[mailbox.Mailbox]"
def _create_logger(self):
global logger
logger = logging.getLogger('cloud-mdir-sync')
logger.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setFormatter(
logging.Formatter(fmt='%(asctime)s %(levelname)-8s %(message)s',
datefmt='%m-%d %H:%M:%S'))
ch.setLevel(logging.DEBUG)
logger.addHandler(ch)
self.logger = logger
def __init__(self):
self._create_logger()
self.cloud_mboxes = []
self.local_mboxes = []
self.message_db_dir = os.path.expanduser(self.message_db_dir)
self.direct_message = self._direct_message
def load_config(self, fn):
"""The configuration file is a python script that we execute with
capitalized functions of this class injected into it"""
fn = os.path.expanduser(fn)
with open(fn, "r") as F:
pyc = compile(source=F.read(), filename=fn, mode="exec")
g = {"cfg": self}
for k in dir(self):
if k[0].isupper():
g[k] = getattr(self, k)
eval(pyc, g)
@property
def storage_key(self):
"""The storage key is used with fernet to manage the authentication
data, which is stored to disk using symmetric encryption. The
decryption key is keld by the system keyring in some secure storage.
On Linux desktop systems this is likely to be something like
gnome-keyring."""
import keyring
from cryptography.fernet import Fernet
ring = keyring.get_keyring()
res = ring.get_password("cloud_mdir_sync", "storage")
if res is None:
res = Fernet.generate_key()
ring.set_password("cloud_mdir_sync", "storage", res)
return res
def all_mboxes(self):
return itertools.chain(self.local_mboxes, self.cloud_mboxes)
def Office365_Account(self, user=None, tenant="common"):
"""Define an Office365 account credential. If user is left as None
then the browser will prompt for the user and the choice will be
cached. To lock the account to a single tenant specify the Azure
Directory name, ie 'contoso.onmicrosoft.com', or the GUID."""
return (user,tenant)
def Office365(self, mailbox, account):
"""Create a cloud mailbox for Office365. Mailbox is the name of O365
mailbox to use, account should be the result of Office365_Account"""
from .office365 import O365Mailbox
self.cloud_mboxes.append(O365Mailbox(mailbox, user=account[0],
tenant=account[1]))
return self.cloud_mboxes[-1]
def GMail_Account(self, user):
"""Define a GMail account credential. The user must be specified as a
fully qualified Google Account email address. This supports both
consumer GMail accounts, and accounts linked to a G-Suite account."""
return (user,)
def GMail(self, label, account):
"""Create a cloud mailbox for Office365. Mailbox is the name of O365
mailbox to use, account should be the result of Office365_Account"""
from .gmail import GMailMailbox
self.cloud_mboxes.append(GMailMailbox(label, user=account[0]))
return self.cloud_mboxes[-1]
def MailDir(self, directory):
"""Create a local maildir to hold messages"""
from .maildir import MailDirMailbox
self.local_mboxes.append(MailDirMailbox(directory))
return self.local_mboxes[-1]
def _direct_message(self, msg):
return self.local_mboxes[0]
|
