feat: 账号登陆

mail-web.py

@@ -1,16 +1,4 @@
 from web import HuamMailFlask
-from web.configure import configure
 
-import os
-import logging
-
-env_dict = os.environ
-huan_mail_conf = env_dict.get("HUAN_MAIL_CONF")
-if huan_mail_conf is None:
-    logging.info("Configure file ./etc/conf.json")
-    configure("./etc/conf.json")
-else:
-    logging.info(f"Configure file {huan_mail_conf}")
-    configure(huan_mail_conf)
 
 app = HuamMailFlask(__name__)

mailbox/imap.py

@@ -101,8 +101,23 @@ class Imap:
     def mailbox(self) -> List[Mail]:
         return sorted(self.__mailbox.values(), reverse=True)
 
+    @mailbox.setter
+    def mailbox(self, args: dict):
+        if type(args) is dict and len(args) == 0:
+            self.__mailbox = {}
+        else:
+            raise ValueError
+
     def add_mail(self, num: str, data: bytes):
         self.__mailbox[num] = Mail(num, data)
 
     def get_mail(self, num: str):
         return self.__mailbox.get(num, None)
+
+    def check_login(self):
+        try:
+            self.connect()
+            self.disconnect()
+        except imaplib.IMAP4.error:
+            return False
+        return True

sender/smtp.py

@@ -26,3 +26,20 @@ class Sender:
                         [i[1] for i in msg.to_addr + msg.cc_addr + msg.bcc_addr],
                         msg.as_string())
         server.quit()
+
+    def check_login(self):
+        if self.ssl:
+            server = smtplib.SMTP_SSL(self.host, self.port)
+        else:
+            server = smtplib.SMTP(self.host, self.port)
+        server.set_debuglevel(self.debug)
+        if self.start_ssl:
+            server.starttls()
+
+        try:
+            server.login(self.user, self.passwd)
+            server.quit()
+        except (smtplib.SMTPHeloError, smtplib.SMTPAuthenticationError,
+                smtplib.SMTPNotSupportedError, smtplib.SMTPException):
+            return False
+        return True

templates/auth/login.html

@@ -2,5 +2,15 @@
 {% block title %} 登录 {% endblock %}
 
 {% block content %}
+    <div class="container">
+        <form method="post" action="{{ url_for("auth.login_page") }}" class="was-validated">
+            {{ form.hidden_tag() }}
+            {{ render_field(form.username) }}
+            {{ render_field(form.passwd) }}
 
+            <div class="text-end">
+                {{ form.submit(class='btn btn-success me-2') }}
+            </div>
+        </form>
+    </div>
 {% endblock %}

templates/base.html

@@ -73,7 +73,7 @@
     {% block nav %}
         <div class="container mt-2">
             <a class="h3" href="/" style="text-decoration:none;color:#333;"> {{ conf["WEBSITE_TITLE"] }} </a>
-            <a href="{{ url_for("base.index_page") }}" class="btn btn-outline-danger float-end mx-2"> 退出登录 </a>
+            <a href="{{ url_for("auth.logout_page") }}" class="btn btn-outline-danger float-end mx-2"> 退出登录 </a>
         </div>
     {% endblock %}
 

templates/index/index.html

@@ -1,2 +1,8 @@
 {% extends "base.html" %}
-{% block title %} 首页 {% endblock %}
+{% block title %} 首页 {% endblock %}
+
+{% block content %}
+<div class="container text-center">
+    <h5> 欢迎，{{ current_user.id }} </h5>
+</div>
+{% endblock %}

web/__init__.py

@@ -35,6 +35,9 @@ class HuamMailFlask(Flask):
         from .index import index
         self.register_blueprint(index, url_prefix="/")
 
+        from .auth import auth
+        self.register_blueprint(auth, url_prefix="/auth")
+
     def profile_setting(self):
         if conf["DEBUG_PROFILE"]:
             self.wsgi_app = ProfilerMiddleware(self.wsgi_app, sort_by=("cumtime",))

web/auth.py

@@ -0,0 +1,52 @@
+from flask import Blueprint, render_template, redirect, url_for, flash, request
+from flask_login import login_required, login_user, logout_user, current_user
+from flask_wtf import FlaskForm
+from wtforms import PasswordField, StringField, SubmitField
+from wtforms.validators import DataRequired
+
+from .logger import Logger
+from .user import User
+
+auth = Blueprint("auth", __name__)
+
+
+class LoginForm(FlaskForm):
+    username = StringField("用户名", description="邮箱用户名", validators=[DataRequired("必须填写用户名")])
+    passwd = PasswordField("密码", description="邮箱密码", validators=[DataRequired("必须填写密码")])
+    submit = SubmitField("登录")
+
+
+@auth.route("/login", methods=["GET", "POST"])
+def login_page():
+    if not current_user.is_anonymous:
+        flash("不能重复登录")
+        return redirect(url_for("base.index_page"))
+
+    form = LoginForm()
+    if form.validate_on_submit():
+        user = User(form.username.data, form.passwd.data)
+        if user.check_login():
+            login_user(user, remember=True)
+            flash("登陆成功")
+            next_page = request.args.get("next")
+            if next_page is None or not next_page.startswith('/'):
+                next_page = url_for('base.index_page')
+            Logger.print_user_opt_success_log(f"login {form.username.data}")
+            return redirect(next_page)
+        else:
+            flash("账号验证失败")
+            Logger.print_user_opt_fail_log(f"login {form.username.data}")
+            return redirect(url_for("auth.login_page"))
+    Logger.print_load_page_log("login")
+    return render_template("auth/login.html", form=form)
+
+
+
+
+@auth.route("/logout")
+@login_required
+def logout_page():
+    logout_user()
+    flash("退出登录成功")
+    Logger.print_user_opt_success_log(f"logout")
+    return redirect(url_for("auth.login_page"))

web/configure.py

@@ -13,11 +13,15 @@ conf: Dict[str, any] = {
     "IMAP_PORT": 143,
     "IMAP_SSL": False,
     "IMAP_START_SSL": False,
+    "IMAP_USERNAME": "{0}",
+    "IMAP_PASSWD": "{0}",
 
     "SMTP_HOST": "localhost",
     "SMTP_PORT": 25,
     "SMTP_SSL": False,
     "SMTP_START_SSL": False,
+    "SMTP_USERNAME": "{0}",
+    "SMTP_PASSWD": "{0}",
 
     "REDIS_HOST": "localhost",
     "REDIS_PORT": 6379,
@@ -51,3 +55,13 @@ def configure(conf_file: str, encoding="utf-8"):
                              "error": logging.ERROR}.get(conf["LOG_LEVEL"])
     if len(conf["LOG_HOME"]) > 0:
         os.makedirs(conf["LOG_HOME"], exist_ok=True)
+
+
+env_dict = os.environ
+huan_mail_conf = env_dict.get("HUAN_MAIL_CONF")
+if huan_mail_conf is None:
+    logging.info("Configure file ./etc/conf.json")
+    configure("./etc/conf.json")
+else:
+    logging.info(f"Configure file {huan_mail_conf}")
+    configure(huan_mail_conf)

web/db.py

@@ -0,0 +1,28 @@
+from redis import StrictRedis
+import logging
+import logging.handlers
+import os
+
+from .configure import conf
+
+
+class RedisDB(StrictRedis):
+    def __init__(self, host, port, username, passwd, db):
+        super().__init__(host=host, port=port, username=username, password=passwd, db=db, decode_responses=True)
+
+        # redis是线程安全的
+
+        self.logger = logging.getLogger("main.database")
+        self.logger.setLevel(conf["LOG_LEVEL"])
+        if len(conf["LOG_HOME"]) > 0:
+            handle = logging.handlers.TimedRotatingFileHandler(
+                os.path.join(conf["LOG_HOME"], f"redis-{username}@{host}.log"), backupCount=10)
+            handle.setFormatter(logging.Formatter(conf["LOG_FORMAT"]))
+            self.logger.addHandler(handle)
+
+
+redis = RedisDB(host=conf["REDIS_HOST"],
+                port=conf["REDIS_PORT"],
+                username=conf["REDIS_NAME"],
+                passwd=conf["REDIS_PASSWD"],
+                db=conf["REDIS_DATABASE"])

web/index.py

@@ -1,9 +1,13 @@
 from flask import Blueprint, render_template
+from flask_login import login_required
+from .logger import Logger
 
 
 index = Blueprint("base", __name__)
 
 
 @index.route("/")
+@login_required
 def index_page():
+    Logger.print_load_page_log("index")
     return render_template("index/index.html")

web/logger.py

@@ -3,57 +3,47 @@ from flask_login import current_user
 
 
 class Logger:
-    @staticmethod
-    def __get_log_request_info():
-        return (f"user: '{current_user.email}' "
-                f"url: '{request.url}' blueprint: '{request.blueprint}' "
-                f"args: {request.args} form: {request.form} "
-                f"accept_encodings: '{request.accept_encodings}' "
-                f"accept_charsets: '{request.accept_charsets}' "
-                f"accept_mimetypes: '{request.accept_mimetypes}' "
-                f"accept_languages: '{request.accept_languages}'")
-
     @staticmethod
     def print_load_page_log(page: str):
         current_app.logger.debug(
-            f"[{request.method}] Load - '{page}' " + Logger.__get_log_request_info())
+            f"[{request.method}] Load - '{page}' ")
 
     @staticmethod
     def print_form_error_log(opt: str):
         current_app.logger.warning(
-            f"[{request.method}] '{opt}' - Bad form " + Logger.__get_log_request_info())
+            f"[{request.method}] '{opt}' - Bad form ")
 
     @staticmethod
     def print_sys_opt_fail_log(opt: str):
         current_app.logger.error(
-            f"[{request.method}] System {opt} - fail " + Logger.__get_log_request_info())
+            f"[{request.method}] System {opt} - fail ")
 
     @staticmethod
     def print_sys_opt_success_log(opt: str):
         current_app.logger.warning(
-            f"[{request.method}] System {opt} - success " + Logger.__get_log_request_info())
+            f"[{request.method}] System {opt} - success ")
 
     @staticmethod
     def print_user_opt_fail_log(opt: str):
         current_app.logger.debug(
-            f"[{request.method}] User {opt} - fail " + Logger.__get_log_request_info())
+            f"[{request.method}] User {opt} - fail ")
 
     @staticmethod
     def print_user_opt_success_log(opt: str):
         current_app.logger.debug(
-            f"[{request.method}] User {opt} - success " + Logger.__get_log_request_info())
+            f"[{request.method}] User {opt} - success ")
 
     @staticmethod
     def print_user_opt_error_log(opt: str):
         current_app.logger.warning(
-            f"[{request.method}] User {opt} - system fail " + Logger.__get_log_request_info())
+            f"[{request.method}] User {opt} - system fail ")
 
     @staticmethod
     def print_import_user_opt_success_log(opt: str):
         current_app.logger.info(
-            f"[{request.method}] User {opt} - success " + Logger.__get_log_request_info())
+            f"[{request.method}] User {opt} - success ")
 
     @staticmethod
     def print_user_not_allow_opt_log(opt: str):
         current_app.logger.info(
-            f"[{request.method}] User '{opt}' - reject " + Logger.__get_log_request_info())
+            f"[{request.method}] User '{opt}' - reject ")

web/login.py

@@ -1,4 +1,5 @@
 from flask_login import LoginManager
+from .user import User
 
 
 login = LoginManager()
@@ -6,5 +7,8 @@ login.login_view = "auth.login_page"
 
 
 @login.user_loader
-def user_loader(user_id: int):
+def user_loader(username):
+    user = User(username)
+    if user.check_login():
+        return user
     return None

web/user.py

@@ -0,0 +1,43 @@
+from sender.smtp import Sender
+from mailbox.imap import Imap
+from .db import redis
+from .configure import conf
+
+from flask_login import UserMixin
+
+
+class User(UserMixin):
+    def __init__(self, username, passwd=None):
+        self.id = username
+        self.username = username
+
+        if passwd:
+            redis.hmset(f"user:{username}", {"passwd": passwd})
+
+    def check_login(self):
+        imap = Imap(user=conf["IMAP_USERNAME"].format(self.username),
+                    passwd=conf["IMAP_PASSWD"].format(self.passwd),
+                    host=conf["IMAP_HOST"],
+                    port=conf["IMAP_PORT"],
+                    ssl=conf["IMAP_SSL"],
+                    start_ssl=conf["IMAP_START_SSL"])
+
+        sender = Sender(user=conf["SMTP_USERNAME"].format(self.username),
+                        passwd=conf["SMTP_PASSWD"].format(self.passwd),
+                        host=conf["SMTP_HOST"],
+                        port=conf["SMTP_PORT"],
+                        ssl=conf["SMTP_SSL"],
+                        start_ssl=conf["SMTP_START_SSL"],
+                        debug=False)
+
+        if not imap.check_login():
+            return False
+        return sender.check_login()
+
+    @property
+    def info(self):
+        return redis.hgetall(f"user:{self.username}")
+
+    @property
+    def passwd(self):
+        return self.info.get("passwd", "123456789")