mediagoblin/decorators.py

   1 # GNU MediaGoblin -- federated, autonomous media hosting
   2 # Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17 from functools import wraps
  18
  19 from urlparse import urljoin
  20 from werkzeug.exceptions import Forbidden, NotFound
  21 from oauthlib.oauth1 import ResourceEndpoint
  22
  23 from mediagoblin import mg_globals as mgg
  24 from mediagoblin import messages
  25 from mediagoblin.db.models import MediaEntry, User, MediaComment, AccessToken
  26 from mediagoblin.tools.response import (
  27     redirect, render_404,
  28     render_user_banned, json_response)
  29 from mediagoblin.tools.translate import pass_to_ugettext as _
  30
  31 from mediagoblin.oauth.tools.request import decode_authorization_header
  32 from mediagoblin.oauth.oauth import GMGRequestValidator
  33
  34
  35 def user_not_banned(controller):
  36     """
  37     Requires that the user has not been banned. Otherwise redirects to the page
  38     explaining why they have been banned
  39     """
  40     @wraps(controller)
  41     def wrapper(request, *args, **kwargs):
  42         if request.user:
  43             if request.user.is_banned():
  44                 return render_user_banned(request)
  45         return controller(request, *args, **kwargs)
  46
  47     return wrapper
  48
  49
  50 def require_active_login(controller):
  51     """
  52     Require an active login from the user. If the user is banned, redirects to
  53     the "You are Banned" page.
  54     """
  55     @wraps(controller)
  56     @user_not_banned
  57     def new_controller_func(request, *args, **kwargs):
  58         if request.user and \
  59                 not request.user.has_privilege(u'active'):
  60             return redirect(
  61                 request, 'mediagoblin.user_pages.user_home',
  62                 user=request.user.username)
  63         elif not request.user or not request.user.has_privilege(u'active'):
  64             next_url = urljoin(
  65                     request.urlgen('mediagoblin.auth.login',
  66                         qualified=True),
  67                     request.url)
  68
  69             return redirect(request, 'mediagoblin.auth.login',
  70                             next=next_url)
  71
  72         return controller(request, *args, **kwargs)
  73
  74     return new_controller_func
  75
  76
  77 def user_has_privilege(privilege_name, allow_admin=True):
  78     """
  79     Requires that a user have a particular privilege in order to access a page.
  80     In order to require that a user have multiple privileges, use this
  81     decorator twice on the same view. This decorator also makes sure that the
  82     user is not banned, or else it redirects them to the "You are Banned" page.
  83
  84         :param privilege_name       A unicode object that is that represents
  85                                         the privilege object. This object is
  86                                         the name of the privilege, as assigned
  87                                         in the Privilege.privilege_name column
  88
  89         :param allow_admin          If this is true then if the user is an admin
  90                                     it will allow the user even if the user doesn't
  91                                     have the privilage given in privilage_name.
  92     """
  93
  94     def user_has_privilege_decorator(controller):
  95         @wraps(controller)
  96         @require_active_login
  97         def wrapper(request, *args, **kwargs):
  98             if not request.user.has_privilege(privilege_name, allow_admin):
  99                 raise Forbidden()
 100
 101             return controller(request, *args, **kwargs)
 102
 103         return wrapper
 104     return user_has_privilege_decorator
 105
 106
 107 def active_user_from_url(controller):
 108     """Retrieve User() from <user> URL pattern and pass in as url_user=...
 109
 110     Returns a 404 if no such active user has been found"""
 111     @wraps(controller)
 112     def wrapper(request, *args, **kwargs):
 113         user = User.query.filter_by(username=request.matchdict['user']).first()
 114         if user is None:
 115             return render_404(request)
 116
 117         return controller(request, *args, url_user=user, **kwargs)
 118
 119     return wrapper
 120
 121
 122 def user_may_delete_media(controller):
 123     """
 124     Require user ownership of the MediaEntry to delete.
 125     """
 126     @wraps(controller)
 127     def wrapper(request, *args, **kwargs):
 128         uploader_id = kwargs['media'].uploader
 129         if not (request.user.has_privilege(u'admin') or
 130                 request.user.id == uploader_id):
 131             raise Forbidden()
 132
 133         return controller(request, *args, **kwargs)
 134
 135     return wrapper
 136
 137
 138 def user_may_alter_collection(controller):
 139     """
 140     Require user ownership of the Collection to modify.
 141     """
 142     @wraps(controller)
 143     def wrapper(request, *args, **kwargs):
 144         creator_id = request.db.User.query.filter_by(
 145             username=request.matchdict['user']).first().id
 146         if not (request.user.has_privilege(u'admin') or
 147                 request.user.id == creator_id):
 148             raise Forbidden()
 149
 150         return controller(request, *args, **kwargs)
 151
 152     return wrapper
 153
 154
 155 def uses_pagination(controller):
 156     """
 157     Check request GET 'page' key for wrong values
 158     """
 159     @wraps(controller)
 160     def wrapper(request, *args, **kwargs):
 161         try:
 162             page = int(request.GET.get('page', 1))
 163             if page < 0:
 164                 return render_404(request)
 165         except ValueError:
 166             return render_404(request)
 167
 168         return controller(request, page=page, *args, **kwargs)
 169
 170     return wrapper
 171
 172
 173 def get_user_media_entry(controller):
 174     """
 175     Pass in a MediaEntry based off of a url component
 176     """
 177     @wraps(controller)
 178     def wrapper(request, *args, **kwargs):
 179         user = User.query.filter_by(username=request.matchdict['user']).first()
 180         if not user:
 181             raise NotFound()
 182
 183         media = None
 184
 185         # might not be a slug, might be an id, but whatever
 186         media_slug = request.matchdict['media']
 187
 188         # if it starts with id: it actually isn't a slug, it's an id.
 189         if media_slug.startswith(u'id:'):
 190             try:
 191                 media = MediaEntry.query.filter_by(
 192                     id=int(media_slug[3:]),
 193                     state=u'processed',
 194                     uploader=user.id).first()
 195             except ValueError:
 196                 raise NotFound()
 197         else:
 198             # no magical id: stuff?  It's a slug!
 199             media = MediaEntry.query.filter_by(
 200                 slug=media_slug,
 201                 state=u'processed',
 202                 uploader=user.id).first()
 203
 204         if not media:
 205             # Didn't find anything?  Okay, 404.
 206             raise NotFound()
 207
 208         return controller(request, media=media, *args, **kwargs)
 209
 210     return wrapper
 211
 212
 213 def get_user_collection(controller):
 214     """
 215     Pass in a Collection based off of a url component
 216     """
 217     @wraps(controller)
 218     def wrapper(request, *args, **kwargs):
 219         user = request.db.User.query.filter_by(
 220             username=request.matchdict['user']).first()
 221
 222         if not user:
 223             return render_404(request)
 224
 225         collection = request.db.Collection.query.filter_by(
 226             slug=request.matchdict['collection'],
 227             creator=user.id).first()
 228
 229         # Still no collection?  Okay, 404.
 230         if not collection:
 231             return render_404(request)
 232
 233         return controller(request, collection=collection, *args, **kwargs)
 234
 235     return wrapper
 236
 237
 238 def get_user_collection_item(controller):
 239     """
 240     Pass in a CollectionItem based off of a url component
 241     """
 242     @wraps(controller)
 243     def wrapper(request, *args, **kwargs):
 244         user = request.db.User.query.filter_by(
 245             username=request.matchdict['user']).first()
 246
 247         if not user:
 248             return render_404(request)
 249
 250         collection_item = request.db.CollectionItem.query.filter_by(
 251             id=request.matchdict['collection_item']).first()
 252
 253         # Still no collection item?  Okay, 404.
 254         if not collection_item:
 255             return render_404(request)
 256
 257         return controller(request, collection_item=collection_item, *args, **kwargs)
 258
 259     return wrapper
 260
 261
 262 def get_media_entry_by_id(controller):
 263     """
 264     Pass in a MediaEntry based off of a url component
 265     """
 266     @wraps(controller)
 267     def wrapper(request, *args, **kwargs):
 268         media = MediaEntry.query.filter_by(
 269                 id=request.matchdict['media_id'],
 270                 state=u'processed').first()
 271         # Still no media?  Okay, 404.
 272         if not media:
 273             return render_404(request)
 274
 275         given_username = request.matchdict.get('user')
 276         if given_username and (given_username != media.get_uploader.username):
 277             return render_404(request)
 278
 279         return controller(request, media=media, *args, **kwargs)
 280
 281     return wrapper
 282
 283
 284 def get_workbench(func):
 285     """Decorator, passing in a workbench as kwarg which is cleaned up afterwards"""
 286
 287     @wraps(func)
 288     def new_func(*args, **kwargs):
 289         with mgg.workbench_manager.create() as workbench:
 290             return func(*args, workbench=workbench, **kwargs)
 291
 292     return new_func
 293
 294
 295 def allow_registration(controller):
 296     """ Decorator for if registration is enabled"""
 297     @wraps(controller)
 298     def wrapper(request, *args, **kwargs):
 299         if not mgg.app_config["allow_registration"]:
 300             messages.add_message(
 301                 request,
 302                 messages.WARNING,
 303                 _('Sorry, registration is disabled on this instance.'))
 304             return redirect(request, "index")
 305
 306         return controller(request, *args, **kwargs)
 307
 308     return wrapper
 309
 310 def allow_reporting(controller):
 311     """ Decorator for if reporting is enabled"""
 312     @wraps(controller)
 313     def wrapper(request, *args, **kwargs):
 314         if not mgg.app_config["allow_reporting"]:
 315             messages.add_message(
 316                 request,
 317                 messages.WARNING,
 318                 _('Sorry, reporting is disabled on this instance.'))
 319             return redirect(request, 'index')
 320
 321         return controller(request, *args, **kwargs)
 322
 323     return wrapper
 324
 325 def get_optional_media_comment_by_id(controller):
 326     """
 327     Pass in a MediaComment based off of a url component. Because of this decor-
 328     -ator's use in filing Media or Comment Reports, it has two valid outcomes.
 329
 330     :returns        The view function being wrapped with kwarg `comment` set to
 331                         the MediaComment who's id is in the URL. If there is a
 332                         comment id in the URL and if it is valid.
 333     :returns        The view function being wrapped with kwarg `comment` set to
 334                         None. If there is no comment id in the URL.
 335     :returns        A 404 Error page, if there is a comment if in the URL and it
 336                         is invalid.
 337     """
 338     @wraps(controller)
 339     def wrapper(request, *args, **kwargs):
 340         if 'comment' in request.matchdict:
 341             comment = MediaComment.query.filter_by(
 342                     id=request.matchdict['comment']).first()
 343
 344             if comment is None:
 345                 return render_404(request)
 346
 347             return controller(request, comment=comment, *args, **kwargs)
 348         else:
 349             return controller(request, comment=None, *args, **kwargs)
 350     return wrapper
 351
 352
 353 def auth_enabled(controller):
 354     """Decorator for if an auth plugin is enabled"""
 355     @wraps(controller)
 356     def wrapper(request, *args, **kwargs):
 357         if not mgg.app.auth:
 358             messages.add_message(
 359                 request,
 360                 messages.WARNING,
 361                 _('Sorry, authentication is disabled on this instance.'))
 362             return redirect(request, 'index')
 363
 364         return controller(request, *args, **kwargs)
 365
 366     return wrapper
 367
 368 def require_admin_or_moderator_login(controller):
 369     """
 370     Require a login from an administrator or a moderator.
 371     """
 372     @wraps(controller)
 373     def new_controller_func(request, *args, **kwargs):
 374         if request.user and \
 375             not (request.user.has_privilege(u'admin')
 376                 or request.user.has_privilege(u'moderator')):
 377
 378             raise Forbidden()
 379         elif not request.user:
 380             next_url = urljoin(
 381                     request.urlgen('mediagoblin.auth.login',
 382                         qualified=True),
 383                     request.url)
 384
 385             return redirect(request, 'mediagoblin.auth.login',
 386                             next=next_url)
 387
 388         return controller(request, *args, **kwargs)
 389
 390     return new_controller_func
 391
 392
 393
 394 def oauth_required(controller):
 395     """ Used to wrap API endpoints where oauth is required """
 396     @wraps(controller)
 397     def wrapper(request, *args, **kwargs):
 398         data = request.headers
 399         authorization = decode_authorization_header(data)
 400
 401         if authorization == dict():
 402             error = "Missing required parameter."
 403             return json_response({"error": error}, status=400)
 404
 405
 406         request_validator = GMGRequestValidator()
 407         resource_endpoint = ResourceEndpoint(request_validator)
 408         valid, r = resource_endpoint.validate_protected_resource_request(
 409                 uri=request.url,
 410                 http_method=request.method,
 411                 body=request.data,
 412                 headers=dict(request.headers),
 413                 )
 414
 415         if not valid:
 416             error = "Invalid oauth prarameter."
 417             return json_response({"error": error}, status=400)
 418
 419         # Fill user if not already
 420         token = authorization[u"oauth_token"]
 421         access_token = AccessToken.query.filter_by(token=token).first()
 422         if access_token is not None and request.user is None:
 423             user_id = access_token.user
 424             request.user = User.query.filter_by(id=user_id).first()
 425
 426         return controller(request, *args, **kwargs)
 427
 428     return wrapper