Adding transfer_validate_token endpoint to projects API

remotes/origin/issue/4795/notification_even_they_are_disabled
Alejandro Alonso 2016-03-15 10:46:28 +01:00
parent 8b998a5b08
commit 97a45c7095
4 changed files with 154 additions and 5 deletions

View File

@ -330,6 +330,14 @@ class ProjectViewSet(LikedResourceMixin, HistoryResourceMixin,
self.check_permissions(request, "tags_colors", project)
return response.Ok(dict(project.tags_colors))
@detail_route(methods=["POST"])
def transfer_validate_token(self, request, pk=None):
project = self.get_object()
self.check_permissions(request, "transfer_validate_token", project)
token = request.DATA.get('token', None)
services.transfer.validate_project_transfer_token(token, project, request.user)
return response.Ok()
@detail_route(methods=["POST"])
def transfer_request(self, request, pk=None):
project = self.get_object()

View File

@ -77,6 +77,7 @@ class ProjectPermission(TaigaResourcePermission):
unwatch_perms = IsAuthenticated() & HasProjectPerm('view_project')
create_template_perms = IsSuperUser()
leave_perms = CanLeaveProject()
transfer_validate_token_perms = IsProjectAdmin()
transfer_request_perms = IsProjectAdmin()
transfer_start_perms = IsMainOwner()
transfer_reject_perms = IsProjectAdmin()

View File

@ -54,10 +54,10 @@ def start_project_transfer(project, user, reason):
email.send()
def _validate_token(token, project_token, user_id):
def validate_project_transfer_token(token, project, user):
signer = signing.TimestampSigner()
if project_token != token:
if project.transfer_token != token:
raise exc.WrongArguments(_("Token is invalid"))
try:
@ -67,12 +67,12 @@ def _validate_token(token, project_token, user_id):
except signing.BadSignature:
raise exc.WrongArguments(_("Token is invalid"))
if str(value) != str(user_id):
if str(value) != str(user.id):
raise exc.WrongArguments(_("Token is invalid"))
def reject_project_transfer(project, user, token, reason):
_validate_token(token, project.transfer_token, user.id)
validate_project_transfer_token(token, project, user)
project.transfer_token = None
project.save()
@ -88,7 +88,7 @@ def reject_project_transfer(project, user, token, reason):
def accept_project_transfer(project, user, token, reason):
_validate_token(token, project.transfer_token, user.id)
validate_project_transfer_token(token, project, user)
old_owner = project.owner

View File

@ -797,6 +797,7 @@ def test_transfer_request_from_admin_member(client):
assert response.status_code == 200
assert len(mail.outbox) == 1
def test_project_transfer_start_to_not_a_membership(client):
user_from = f.UserFactory.create()
project = f.create_project(owner=user_from)
@ -1253,3 +1254,142 @@ def test_project_transfer_accept_from_admin_member_with_valid_token_with_enough_
project = Project.objects.get(pk=project.pk)
assert project.owner.id == user_to.id
assert project.transfer_token is None
def test_project_transfer_validate_token_from_admin_member_without_token(client):
user_from = f.UserFactory.create()
user_to = f.UserFactory.create()
signer = signing.TimestampSigner()
token = signer.sign(user_to.id)
project = f.create_project(owner=user_from, transfer_token=token)
f.MembershipFactory(user=user_from, project=project, is_admin=True)
f.MembershipFactory(user=user_to, project=project, is_admin=True)
client.login(user_to)
url = reverse("projects-transfer-validate-token", kwargs={"pk": project.pk})
data = {}
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
def test_project_transfer_validate_token_from_not_admin_member(client):
user_from = f.UserFactory.create()
user_to = f.UserFactory.create()
signer = signing.TimestampSigner()
token = signer.sign(user_to.id)
project = f.create_project(owner=user_from, transfer_token=token, public_permissions=["view_project"])
f.MembershipFactory(user=user_from, project=project, is_admin=True)
f.MembershipFactory(user=user_to, project=project, is_admin=False)
client.login(user_to)
url = reverse("projects-transfer-validate-token", kwargs={"pk": project.pk})
data = {
"token": token,
}
response = client.json.post(url, json.dumps(data))
assert response.status_code == 403
def test_project_transfer_validate_token_from_admin_member_with_invalid_token(client):
user_from = f.UserFactory.create()
user_to = f.UserFactory.create()
project = f.create_project(owner=user_from, transfer_token="invalid-token")
f.MembershipFactory(user=user_from, project=project, is_admin=True)
f.MembershipFactory(user=user_to, project=project, is_admin=True)
client.login(user_to)
url = reverse("projects-transfer-validate-token", kwargs={"pk": project.pk})
data = {
"token": "invalid-token",
}
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "Token is invalid" == response.data["_error_message"]
def test_project_transfer_validate_token_from_admin_member_with_other_user_token(client):
user_from = f.UserFactory.create()
user_to = f.UserFactory.create()
other_user = f.UserFactory.create()
signer = signing.TimestampSigner()
token = signer.sign(other_user.id)
project = f.create_project(owner=user_from, transfer_token=token)
f.MembershipFactory(user=user_from, project=project, is_admin=True)
f.MembershipFactory(user=user_to, project=project, is_admin=True)
client.login(user_to)
url = reverse("projects-transfer-validate-token", kwargs={"pk": project.pk})
data = {
"token": token,
}
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "Token is invalid" == response.data["_error_message"]
def test_project_transfer_validate_token_from_admin_member_with_expired_token(client):
user_from = f.UserFactory.create()
user_to = f.UserFactory.create()
signer = ExpiredSigner()
token = signer.sign(user_to.id)
project = f.create_project(owner=user_from, transfer_token=token)
f.MembershipFactory(user=user_from, project=project, is_admin=True)
f.MembershipFactory(user=user_to, project=project, is_admin=True)
client.login(user_to)
url = reverse("projects-transfer-validate-token", kwargs={"pk": project.pk})
data = {
"token": token,
}
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "Token has expired" == response.data["_error_message"]
def test_project_transfer_validate_token_from_admin_member_with_valid_token(client):
user_from = f.UserFactory.create()
user_to = f.UserFactory.create(max_private_projects=1)
signer = signing.TimestampSigner()
token = signer.sign(user_to.id)
project = f.create_project(owner=user_from, transfer_token=token, is_private=True)
f.MembershipFactory(user=user_from, project=project, is_admin=True)
f.MembershipFactory(user=user_to, project=project, is_admin=True)
client.login(user_to)
url = reverse("projects-transfer-validate-token", kwargs={"pk": project.pk})
data = {
"token": token,
}
response = client.json.post(url, json.dumps(data))
assert response.status_code == 200