kallithea Changeset - a45191e7c7bb

Changeset - a45191e7c7bb

Parent rev.

Child rev.

[Not reviewed]

beta

0 4 0

Mads Kiilerich - 13 years ago 2013-01-02 13:56:44
madski@unity3d.com

access control: fix owner checks - they were always true

The lambda expressions seems to be left over from something else. They were no
longer executed and thus always evaluated to true.

Some of the functions also failed if they were executed.

4 files changed with 9 insertions and 8 deletions:

rhodecode/controllers/admin/notifications.py

rhodecode/controllers/changeset.py

rhodecode/controllers/pullrequests.py

rhodecode/tests/functional/test_admin_notifications.py

0 comments (0 inline, 0 general)

rhodecode/controllers/admin/notifications.py

➞

Show inline comments

@@ @@ -65,108 +65,108 @@ class NotificationsController(BaseContro @@
                                             filter_=request.GET.getall('type'))
         p = safe_int(request.params.get('page', 1), 1)
         c.notifications = Page(notif, page=p, items_per_page=10)
         c.pull_request_type = Notification.TYPE_PULL_REQUEST
         c.comment_type = [Notification.TYPE_CHANGESET_COMMENT,
                           Notification.TYPE_PULL_REQUEST_COMMENT]
         _current_filter = request.GET.getall('type')
         c.current_filter = 'all'
         if _current_filter == [c.pull_request_type]:
             c.current_filter = 'pull_request'
         elif _current_filter == c.comment_type:
             c.current_filter = 'comment'
         return render('admin/notifications/notifications.html')
     def mark_all_read(self):
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             nm = NotificationModel()
             # mark all read
             nm.mark_all_read_for_user(self.rhodecode_user.user_id,
                                       filter_=request.GET.getall('type'))
             Session().commit()
             c.user = self.rhodecode_user
             notif = nm.get_for_user(self.rhodecode_user.user_id,
                                     filter_=request.GET.getall('type'))
             c.notifications = Page(notif, page=1, items_per_page=10)
             return render('admin/notifications/notifications_data.html')
     def create(self):
         """POST /_admin/notifications: Create a new item"""
         # url('notifications')
     def new(self, format='html'):
         """GET /_admin/notifications/new: Form to create a new item"""
         # url('new_notification')
     def update(self, notification_id):
         """PUT /_admin/notifications/id: Update an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="PUT" />
         # Or using helpers:
         #    h.form(url('notification', notification_id=ID),
         #           method='put')
         # url('notification', notification_id=ID)
         try:
             no = Notification.get(notification_id)
             owner = lambda: (no.notifications_to_users.user.user_id
                              == c.rhodecode_user.user_id)
             owner = all(un.user.user_id == c.rhodecode_user.user_id
                         for un in no.notifications_to_users)
             if h.HasPermissionAny('hg.admin')() or owner:
                     NotificationModel().mark_read(c.rhodecode_user.user_id, no)
                     Session().commit()
                     return 'ok'
         except Exception:
             Session.rollback()
             log.error(traceback.format_exc())
         return 'fail'
     def delete(self, notification_id):
         """DELETE /_admin/notifications/id: Delete an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="DELETE" />
         # Or using helpers:
         #    h.form(url('notification', notification_id=ID),
         #           method='delete')
         # url('notification', notification_id=ID)
         try:
             no = Notification.get(notification_id)
             owner = lambda: (no.notifications_to_users.user.user_id
                              == c.rhodecode_user.user_id)
             owner = all(un.user.user_id == c.rhodecode_user.user_id
                         for un in no.notifications_to_users)
             if h.HasPermissionAny('hg.admin')() or owner:
                     NotificationModel().delete(c.rhodecode_user.user_id, no)
                     Session().commit()
                     return 'ok'
         except Exception:
             Session.rollback()
             log.error(traceback.format_exc())
         return 'fail'
     def show(self, notification_id, format='html'):
         """GET /_admin/notifications/id: Show a specific item"""
         # url('notification', notification_id=ID)
         c.user = self.rhodecode_user
         no = Notification.get(notification_id)
         owner = lambda: (no.notifications_to_users.user.user_id
                          == c.user.user_id)
         owner = all(un.user.user_id == c.rhodecode_user.user_id
                     for un in no.notifications_to_users)
         if no and (h.HasPermissionAny('hg.admin', 'repository.admin')() or owner):
             unotification = NotificationModel()\
                             .get_user_notification(c.user.user_id, no)
             # if this association to user is not valid, we don't want to show
             # this message
             if unotification:
                 if unotification.read is False:
                     unotification.mark_as_read()
                     Session().commit()
                 c.notification = no
                 return render('admin/notifications/show_notification.html')
         return redirect(url('notifications'))
     def edit(self, notification_id, format='html'):
         """GET /_admin/notifications/id/edit: Form to edit an existing item"""
         # url('edit_notification', notification_id=ID)

rhodecode/controllers/changeset.py

➞

Show inline comments

@@ @@ -326,65 +326,65 @@ class ChangesetController(BaseRepoContro @@
+        )
         # get status if set !
         if status and change_status:
             # if latest status was from pull request and it's closed
             # disallow changing status !
             # dont_allow_on_closed_pull_request = True !
             try:
                 ChangesetStatusModel().set_status(
                     c.rhodecode_db_repo.repo_id,
                     status,
                     c.rhodecode_user.user_id,
                     comm,
                     revision=revision,
                     dont_allow_on_closed_pull_request=True
+                )
             except StatusChangeOnClosedPullRequestError:
                 log.error(traceback.format_exc())
                 msg = _('Changing status on a changeset associated with'
                         'a closed pull request is not allowed')
                 h.flash(msg, category='warning')
                 return redirect(h.url('changeset_home', repo_name=repo_name,
                                       revision=revision))
         action_logger(self.rhodecode_user,
                       'user_commented_revision:%s' % revision,
                       c.rhodecode_db_repo, self.ip_addr, self.sa)
         Session().commit()
         if not request.environ.get('HTTP_X_PARTIAL_XHR'):
             return redirect(h.url('changeset_home', repo_name=repo_name,
                                   revision=revision))
         data = {
            'target_id': h.safeid(h.safe_unicode(request.POST.get('f_path'))),
+        }
         if comm:
             c.co = comm
             data.update(comm.get_dict())
             data.update({'rendered_text':
                          render('changeset/changeset_comment_block.html')})
         return data
     @jsonify
     def delete_comment(self, repo_name, comment_id):
         co = ChangesetComment.get(comment_id)
-        owner = lambda: co.author.user_id == c.rhodecode_user.user_id
         owner = co.author.user_id == c.rhodecode_user.user_id
         if h.HasPermissionAny('hg.admin', 'repository.admin')() or owner:
             ChangesetCommentsModel().delete(comment=co)
             Session().commit()
             return True
         else:
             raise HTTPForbidden()
     @jsonify
     def changeset_info(self, repo_name, revision):
         if request.is_xhr:
             try:
                 return c.rhodecode_repo.get_changeset(revision)
             except ChangesetDoesNotExistError, e:
                 return EmptyChangeset(message=str(e))
         else:
             raise HTTPBadRequest()

rhodecode/controllers/pullrequests.py

➞

Show inline comments

@@ @@ -432,55 +432,55 @@ class PullrequestsController(BaseRepoCon @@
         if allowed_to_change_status:
             # get status if set !
             if status and change_status:
                 ChangesetStatusModel().set_status(
                     c.rhodecode_db_repo.repo_id,
                     status,
                     c.rhodecode_user.user_id,
                     comm,
                     pull_request=pull_request_id
+                )
             if request.POST.get('save_close'):
                 if status in ['rejected', 'approved']:
                     PullRequestModel().close_pull_request(pull_request_id)
                     action_logger(self.rhodecode_user,
                               'user_closed_pull_request:%s' % pull_request_id,
                               c.rhodecode_db_repo, self.ip_addr, self.sa)
                 else:
                     h.flash(_('Closing pull request on other statuses than '
                               'rejected or approved forbidden'),
                             category='warning')
         Session().commit()
         if not request.environ.get('HTTP_X_PARTIAL_XHR'):
             return redirect(h.url('pullrequest_show', repo_name=repo_name,
                                   pull_request_id=pull_request_id))
         data = {
            'target_id': h.safeid(h.safe_unicode(request.POST.get('f_path'))),
+        }
         if comm:
             c.co = comm
             data.update(comm.get_dict())
             data.update({'rendered_text':
                          render('changeset/changeset_comment_block.html')})
         return data
     @NotAnonymous()
     @jsonify
     def delete_comment(self, repo_name, comment_id):
         co = ChangesetComment.get(comment_id)
         if co.pull_request.is_closed():
             #don't allow deleting comments on closed pull request
             raise HTTPForbidden()
-        owner = lambda: co.author.user_id == c.rhodecode_user.user_id
         owner = co.author.user_id == c.rhodecode_user.user_id
         if h.HasPermissionAny('hg.admin', 'repository.admin')() or owner:
             ChangesetCommentsModel().delete(comment=co)
             Session().commit()
             return True
         else:
             raise HTTPForbidden()

rhodecode/tests/functional/test_admin_notifications.py

➞

Show inline comments

@@ @@ -37,69 +37,70 @@ class TestNotificationsController(TestCo @@
 #    def test_index_as_xml(self):
 #        response = self.app.get(url('formatted_notifications', format='xml'))
+#
 #    def test_create(self):
 #        response = self.app.post(url('notifications'))
+#
 #    def test_new(self):
 #        response = self.app.get(url('new_notification'))
+#
 #    def test_new_as_xml(self):
 #        response = self.app.get(url('formatted_new_notification', format='xml'))
+#
 #    def test_update(self):
 #        response = self.app.put(url('notification', notification_id=1))
+#
 #    def test_update_browser_fakeout(self):
 #        response = self.app.post(url('notification', notification_id=1), params=dict(_method='put'))
     def test_delete(self):
         self.log_user()
         cur_user = self._get_logged_user()
         u1 = UserModel().create_or_update(username='u1', password='qweqwe',
                                                email='u1@rhodecode.org',
                                                firstname='u1', lastname='u1')
         u2 = UserModel().create_or_update(username='u2', password='qweqwe',
                                                email='u2@rhodecode.org',
                                                firstname='u2', lastname='u2')
         # make notifications
         notification = NotificationModel().create(created_by=cur_user,
                                                   subject=u'test',
                                                   body=u'hi there',
                                                   recipients=[cur_user, u1, u2])
         self.Session().commit()
         u1 = User.get(u1.user_id)
         u2 = User.get(u2.user_id)
         # check DB
         get_notif = lambda un: [x.notification for x in un]
         self.assertEqual(get_notif(cur_user.notifications), [notification])
         self.assertEqual(get_notif(u1.notifications), [notification])
         self.assertEqual(get_notif(u2.notifications), [notification])
         cur_usr_id = cur_user.user_id
         response = self.app.delete(url('notification',
                                        notification_id=
                                        notification.notification_id))
         self.assertEqual(response.body, 'ok')
         cur_user = User.get(cur_usr_id)
         self.assertEqual(cur_user.notifications, [])
     def test_show(self):
         self.log_user()
         cur_user = self._get_logged_user()
         u1 = UserModel().create_or_update(username='u1', password='qweqwe',
                                                email='u1@rhodecode.org',
                                                firstname='u1', lastname='u1')
         u2 = UserModel().create_or_update(username='u2', password='qweqwe',
                                                email='u2@rhodecode.org',
                                                firstname='u2', lastname='u2')
         notification = NotificationModel().create(created_by=cur_user,
                                                   subject=u'test',
                                                   body=u'hi there',
                                                   recipients=[cur_user, u1, u2])
         response = self.app.get(url('notification',
                                     notification_id=notification.notification_id))

0 comments (0 inline, 0 general)