51 lines
1.8 KiB
Python
51 lines
1.8 KiB
Python
import base64
|
|
import hashlib
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
from flask import current_app
|
|
from sqlalchemy.exc import OperationalError, ProgrammingError
|
|
from app.extensions import db
|
|
from app.models.base import TimestampMixin
|
|
|
|
|
|
class AppSetting(TimestampMixin, db.Model):
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
key = db.Column(db.String(128), unique=True, nullable=False, index=True)
|
|
value = db.Column(db.Text)
|
|
is_encrypted = db.Column(db.Boolean, default=False, nullable=False)
|
|
|
|
@classmethod
|
|
def _cipher(cls):
|
|
secret = current_app.config.get('APP_MASTER_KEY', current_app.config.get('SECRET_KEY', 'dev')).encode('utf-8')
|
|
digest = hashlib.sha256(secret).digest()
|
|
return Fernet(base64.urlsafe_b64encode(digest))
|
|
|
|
@classmethod
|
|
def get(cls, key, default=None, decrypt=False):
|
|
try:
|
|
item = cls.query.filter_by(key=key).first()
|
|
if not item:
|
|
return default
|
|
if decrypt and item.is_encrypted and item.value:
|
|
try:
|
|
return cls._cipher().decrypt(item.value.encode('utf-8')).decode('utf-8')
|
|
except InvalidToken:
|
|
return default
|
|
return item.value if item.value is not None else default
|
|
except (OperationalError, ProgrammingError):
|
|
return default
|
|
|
|
@classmethod
|
|
def set(cls, key, value, encrypt=False):
|
|
item = cls.query.filter_by(key=key).first()
|
|
if not item:
|
|
item = cls(key=key)
|
|
db.session.add(item)
|
|
item.is_encrypted = encrypt
|
|
if value is None:
|
|
item.value = None
|
|
elif encrypt:
|
|
item.value = cls._cipher().encrypt(str(value).encode('utf-8')).decode('utf-8')
|
|
else:
|
|
item.value = str(value)
|
|
return item
|