Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class Fixture:
def on_request(self):
pass # called when a request arrives
def on_error(self):
pass # called when a request errors
def on_success(self):
pass # called when a request is successful
def transform(self, output): # transforms the output, for example to apply template
return output
class Translator(pluralize.Translator, Fixture):
def on_request(self):
self.select(request.headers.get("Accept-Language", "en"))
def on_success(self):
response.headers["Content-Language"] = self.local.tag
class DAL(pydal.DAL, Fixture):
def on_request(self):
threadsafevariable.ThreadSafeVariable.restore(ICECUBE)
def on_error(self):
self.rollback()
def on_success(self):
self.commit()
def on_success(self):
pass # called when a request is successful
def transform(self, output): # transforms the output, for example to apply template
return output
class Translator(pluralize.Translator, Fixture):
def on_request(self):
self.select(request.headers.get("Accept-Language", "en"))
def on_success(self):
response.headers["Content-Language"] = self.local.tag
class DAL(pydal.DAL, Fixture):
def on_request(self):
threadsafevariable.ThreadSafeVariable.restore(ICECUBE)
def on_error(self):
self.rollback()
def on_success(self):
self.commit()
# make sure some variables in pydal are thread safe
for _ in ["readable", "writable", "default", "update", "requires"]:
setattr(pydal.DAL.Field, _, threadsafevariable.ThreadSafeVariable())
# this global object will be used to store their state to restore it for every http request
ICECUBE = {}
self.commit()
# make sure some variables in pydal are thread safe
for _ in ["readable", "writable", "default", "update", "requires"]:
setattr(pydal.DAL.Field, _, threadsafevariable.ThreadSafeVariable())
# this global object will be used to store their state to restore it for every http request
ICECUBE = {}
#########################################################################################
# The Template Rendered Fixture
#########################################################################################
class Template(Fixture):
cache = Cache(100)
def __init__(self, filename, path=None, delimiters="[[ ]]"):
self.filename = filename
self.path = path
self.delimiters = delimiters
@staticmethod
def reader(filename):
"""Cached file reader, only reads template if it has changed"""
def raw_read():
with open(filename, encoding="utf8") as stream:
return stream.read()
import datetime
import hashlib
import urllib
import uuid
from py4web import redirect, request, response, abort, URL, action
from py4web.core import Fixture, Template
from pydal.validators import IS_EMAIL, CRYPT, IS_NOT_EMPTY, IS_NOT_IN_DB
class AuthEnforcer(Fixture):
"""
Base fixtures that checks if a condition is met
if not redirects to a different pages or returns HTTP 403
"""
def __init__(self, auth, condition=None):
self.__prerequisites__ = [auth]
self.auth = auth
self.condition = condition
def abort_or_rediect(self, page):
"""
return HTTP 403 if content_type is applicaitons/json
else redirects to page"""
if request.content_type == "application/json":
output = yatl.render(
Template.reader(filename),
path=path,
context=context,
delimiters=self.delimiters,
reader=Template.reader,
)
return output
#########################################################################################
# The Session Fixture
#########################################################################################
class Session(Fixture):
# All apps share the same default secret if not specified. important for _dashboard reload
SECRET = None
def __init__(
self,
secret=None,
expiration=None,
algorithm="HS256",
storage=None,
same_site="Lax",
):
"""
secret is the shared key used to encrypt the session (using algorithm)
expiration is in seconds
(optional) storage must have a get(key) and set(key,value,expiration) methods
from py4web.core import Reloader, dumps, ErrorStorage, Session, Fixture
from pydal.validators import CRYPT
from yatl.helpers import BEAUTIFY
from .utils import *
MODE = os.environ.get("PY4WEB_DASHBOARD_MODE", "none")
FOLDER = os.environ["PY4WEB_APPS_FOLDER"]
APP_FOLDER = os.path.dirname(__file__)
T_FOLDER = os.path.join(APP_FOLDER, "translations")
T = Translator(T_FOLDER)
error_storage = ErrorStorage()
db = error_storage.db
session = Session()
class Logged(Fixture):
def __init__(self, session):
self.__prerequisites__ = [session]
self.session = session
def on_request(self):
user = self.session.get("user")
if not user or not user.get("id"):
abort(403)
session_secured = action.uses(Logged(session))
if MODE in ("demo", "readonly", "full"):
@action("index")
@action.uses("index.html", session, T)
else redirects to page"""
if request.content_type == "application/json":
abort(403)
redirect(URL(self.auth.route, page))
def on_request(self):
"""check that we have a user in the session and
the condition is met"""
user = self.auth.session.get("user")
if not user or not user.get("id"):
self.abort_or_rediect("login")
if callable(self.condition) and not self.condition(user):
self.abort_or_rediect("not-authorized")
class Auth(Fixture):
messages = {
"verify_email": {
"subject": "Confirm email",
"body": "Welcome {first_name}, click {link} to confirm your email",
},
"reset_password": {
"subject": "Password reset",
"body": "Hello {first_name}, click {link} to change password",
},
"unsubscribe": {
"subject": "Unsubscribe confirmation",
"body": "By {first_name}, you have been erased from our system",
},
}