Error 500
An error occurred trying to process your request.
The system administrator has been notified to fix the problem.
Sometimes, a database update needs to have been run, or you
need to update your browser's local version of the application.
Please return to the home page to run and
receive any updates.
"""
return web.internalerror(s)
def emergency_notice():
"""
Returns emergency notice text if any is set.
"""
if EMERGENCY_NOTICE != "":
if os.path.exists(EMERGENCY_NOTICE):
s = asm3.utils.read_text_file(EMERGENCY_NOTICE)
return s
return ""
def generate_routes():
""" Extract the url property from all classes and construct the route list """
g = globals().copy()
for name, obj in g.items():
try:
url = getattr(obj, "url")
if url != "" and name != "web":
if not url.startswith("/"): url = "/%s" % url
routes.append(url)
routes.append(name)
except:
pass # Ignore objects that don't have url attributes
return routes
class ASMEndpoint(object):
""" Base class for ASM endpoints """
url = "" # The route/url to this target
get_permissions = ( ) # List of permissions needed to GET
post_permissions = ( ) # List of permissions needed to POST
check_logged_in = True # Check whether we have a valid login
user_activity = True # Hitting this endpoint qualifies as user activity
use_web_input = True # Unpack values with webpy's web.input()
login_url = "login" # The url to go to if not logged in
data = None # Request data posted to this endpoint as bytes or str if data_encoding is set
data_encoding = None # codec to use for decoding of posted data to str (None to not decode)
def _params(self):
l = session.locale
if l is None:
l = LOCALE
post = asm3.utils.PostedData({}, l)
try:
if self.use_web_input: post = asm3.utils.PostedData(web.input(filechooser = {}), l)
self.data = web.data()
if self.data_encoding: self.data = asm3.utils.bytes2str(self.data, encoding=self.data_encoding)
except Exception as err:
asm3.al.error("Failed unpacking params: %s" % str(err), "ASMEndpoint._params", session.dbo, sys.exc_info())
return web.utils.storage( data=self.data, post=post, dbo=session.dbo, locale=l, user=session.user, session=session, \
siteid = session.siteid, locationfilter = session.locationfilter, staffid = session.staffid,
visibleanimalids = session.visibleanimalids )
def check(self, permissions):
""" Check logged in and permissions (which can be a single permission string or a list/tuple) """
if self.check_logged_in:
self.check_loggedin(session, web, self.login_url)
if isinstance(permissions, str):
asm3.users.check_permission(session, permissions)
else:
for p in permissions:
asm3.users.check_permission(session, p)
def checkb(self, permissions):
""" Check logged in and a single permission, returning a boolean """
if self.check_logged_in:
self.check_loggedin(session, web, self.login_url)
return asm3.users.check_permission_bool(session, permissions)
def check_animal(self, a):
""" Checks whether the animal we're about to look at is viewable by the user """
if not asm3.animal.is_animal_in_location_filter(a, session.locationfilter, session.siteid, session.visibleanimalids):
raise asm3.utils.ASMPermissionError("animal not in location filter/site")
def check_locked_db(self):
if session.dbo and session.dbo.locked:
l = session.locale
raise asm3.utils.ASMPermissionError(_("This database is locked.", l))
def check_loggedin(self, session, web, loginpage = "login"):
"""
Checks if we have a logged in user and if not, redirects to
the login page
"""
if not self.is_loggedin(session):
path = web.ctx.path
if path.startswith("/"): path = path[1:]
query = str(web.ctx.query)
raise web.seeother("%s/%s?target=%s%s" % (BASE_URL, loginpage, path, query))
elif self.user_activity:
# update the last user activity if logged in
asm3.users.update_user_activity(session.dbo, session.user)
def content(self, o):
""" Virtual function: override to get the content """
return ""
def cache_control(self, client_ttl = 0, cache_ttl = 0):
""" Sends a cache control header.
client_ttl: The max-age to send for the client
cache_ttl: The s-maxage to send for an edge cache
"""
if client_ttl == 0 and cache_ttl == 0:
self.header("Cache-Control", "no-cache, no-store, must-revalidate, max-age=0")
elif client_ttl > 0 and cache_ttl == 0:
self.header("Cache-Control", "public, max-age=%s" % client_ttl)
else:
self.header("Cache-Control", "public, max-age=%s, s-maxage=%s" % (client_ttl, cache_ttl))
def content_type(self, ct):
""" Sends a content-type header """
self.header("Content-Type", ct)
def data_param(self, p):
""" Returns a URL encoded parameter from the data stream.
This is useful for some services where they send data in
odd encodings (eg: PayPal use cp1252) and we can't use
web.input, which assumes utf-8 """
for b in self.data.split("&"):
if b.startswith(p):
return b.split("=")[1]
return ""
def get_cookie(self, s):
""" Returns the value of cookie s. Returns None if it does not exist. """
try:
return web.cookies().get(s)
except:
return None
def set_cookie(self, name, value, ttl):
""" Sets a cookie value """
web.setcookie(name, value, expires=ttl, secure=SESSION_SECURE_COOKIE, httponly=True)
def header(self, key, value):
""" Set the response header key to value """
web.header(key, value)
def is_loggedin(self, session):
"""
Returns true if the user is logged in
"""
return "user" in session and session.user is not None
def notfound(self):
""" Returns a 404 """
raise web.notfound()
def post_all(self, o):
""" Virtual function: override to handle postback """
return ""
def query(self):
""" Returns the request query string """
return web.ctx.query
def redirect(self, route):
""" Redirect to another route
Uses BASE_URL if a relative route is given to help CDNs. """
if not route.startswith("http"): route = "%s/%s" % (BASE_URL, route)
raise web.seeother(route)
def referer(self):
""" Returns the referer request header """
return web.ctx.env.get("HTTP_REFERER", "")
def reload_config(self):
""" Reloads items in the session based on database values, invalidates config.js so client reloads it """
asm3.users.update_session(session)
def remote_ip(self):
""" Gets the IP address of the requester, taking account of reverse proxies """
remoteip = web.ctx['ip']
if "HTTP_X_FORWARDED_FOR" in web.ctx.env:
xf = web.ctx.env["HTTP_X_FORWARDED_FOR"]
if xf is not None and str(xf).strip() != "":
remoteip = xf
return remoteip
def user_agent(self):
""" Returns the user agent request header """
return web.ctx.env.get("HTTP_USER_AGENT", "")
def GET(self):
self.check(self.get_permissions)
return self.content(self._params())
def POST(self):
""" Handle a POST, deal with permissions and locked databases """
if self.check_logged_in:
self.check_locked_db()
self.check(self.post_permissions)
o = self._params()
mode = o.post["mode"]
if mode == "":
return self.post_all(o)
else:
# Mode has been supplied, call post_mode
return getattr(self.__class__, "post_%s" % mode)(self, o)
class GeneratorEndpoint(ASMEndpoint):
"""Base class for endpoints that use generators for their content """
def GET(self):
self.check(self.get_permissions)
if LARGE_FILES_CHUNKED:
self.header("Transfer-Encoding", "chunked")
for x in self.content(self._params()):
yield x
class JSONEndpoint(ASMEndpoint):
""" Base class for ASM endpoints that return JSON """
js_module = "" # The javascript module to start (can be omitted if same as url)
url = "" # The route/url to this target
def controller(self, o):
""" Virtual function to be overridden - return controller as a dict """
return {}
def GET(self):
""" Handle a GET, deal with permissions, session and JSON responses """
self.check(self.get_permissions)
o = self._params()
c = self.controller(o)
self.cache_control(0)
if self.js_module == "":
self.js_module = self.url
if not o.post["json"] == "true":
self.content_type("text/html")
self.header("X-Frame-Options", "SAMEORIGIN")
self.header("X-Content-Type-Options", "nosniff")
self.header("X-XSS-Protection", "1; mode=block")
self.header("Referrer-Policy", "same-origin")
self.header("Strict-Transport-Security", "max-age=%s" % CACHE_ONE_MONTH)
nonce = asm3.utils.uuid_str()
# CSP is not applied to users of the mobile app as we still have users with
# older iPads on iOS/Safari 9 that only supports CSP1
if CONTENT_SECURITY_POLICY != "" and not o.session.mobileapp:
self.header("Content-Security-Policy", CONTENT_SECURITY_POLICY % { "nonce": nonce })
return "%(header)s\n" \
"\n