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
|
from datetime import datetime, timedelta
from flask import current_app, g
from flask_login import UserMixin, current_user
from .model.read_mails import QMailAuthuser
EXPIRATION_SEC = 60 * 60 * 25
class JWebmailUser(UserMixin):
def __init__(self, mail_addr, password):
self.id = mail_addr
self.password = password
class RedisTimeoutSession:
def __init__(self, username, passwd, timeout, port=6379):
import redis
self.timeout = timeout
self.conn = redis.Redis(
host="localhost",
port=port,
decode_responses=True,
protocol=3,
username=username,
password=passwd,
)
def set(self, key, value):
self.conn.setex(f"jwm:user:{key}", self.timeout, value)
def get(self, key):
return self.conn.getex(f"jwm:user:{key}", self.timeout)
class MysqlTimeoutSession:
def __init__(self, username, passwd, timeout, port=3306):
import mysql.connector
self.timeout = timeout
self.conn = mysql.connector.connect(
host="localhost",
port=port,
username=username,
password=passwd,
database="jwebmaildb1",
)
def set(self, key, value):
timeout = datetime.now() + timedelta(seconds=self.timeout)
cur = self.conn.cursor()
cur.execute("DELETE FROM session WHERE user = %s", [key])
cur.execute("INSERT INTO session VALUES (%s, %s, %s)", [key, value, timeout])
self.conn.commit()
cur.close()
def get(self, key):
cur = self.conn.cursor()
cur.execute("DELETE FROM session WHERE timeout < NOW()")
cur.execute("SELECT password FROM session WHERE user = %s", [key])
row = cur.fetchone()
self.conn.commit()
cur.close()
if row is None:
return None
else:
return row[0]
def select_timeout_session():
session_type = current_app.config["JWEBMAIL"]["READ_MAILS"]["SESSION_TYPE"]
if session_type == "REDIS":
return RedisTimeoutSession
elif session_type == "MYSQL":
return MysqlTimeoutSession
else:
raise ValueError(f"unknown session_type {session_type!r}")
def build_qma(username, password):
authenticator = current_app.config["JWEBMAIL"]["READ_MAILS"]["AUTHENTICATOR"]
backend = current_app.config["JWEBMAIL"]["READ_MAILS"]["BACKEND"]
mailbox = current_app.config["JWEBMAIL"]["READ_MAILS"]["MAILBOX"]
mailbox_user = current_app.config["JWEBMAIL"]["READ_MAILS"]["MAILBOX_USER"]
return QMailAuthuser(
username, password, backend, mailbox, mailbox_user, authenticator
)
def login(username, password):
return build_qma(username, password).verify_user()
def add_user(user: JWebmailUser):
passwd = current_app.config["JWEBMAIL"]["READ_MAILS"]["SESSION_STORE_PASSWD"]
r = select_timeout_session()("jwebmail", passwd, EXPIRATION_SEC)
r.set(user.get_id(), user.password)
def load_user(username: str) -> JWebmailUser:
ss_password = current_app.config["JWEBMAIL"]["READ_MAILS"]["SESSION_STORE_PASSWD"]
r = select_timeout_session()("jwebmail", ss_password, EXPIRATION_SEC)
passwd = r.get(username)
if passwd is None:
return None
return JWebmailUser(username, passwd)
def get_read_mails_logged_in():
if "read_mails" in g:
return g.read_mails
qma = build_qma(current_user.get_id(), current_user.password)
g.read_mails = qma
return qma
|
