mirror of
https://github.com/peeringdb/peeringdb.git
synced 2024-05-11 05:55:09 +00:00
aed211418d
* Update 0006_network_allow_ixp_update.py fixes 'Sepcifies' typo in 0006_network_allow_ixp_update.py * Update 0029_auto_20200401_1006.py fixes 'Sepcifies' typo in 0029_auto_20200401_1006.py * Update models.py fixes 'Sepcifies' typo in models.py * Update org_admin_views.py fixes 'afiliation' typos in org_admin_views.py * Update serializers.py fix use of "it's" when it should instead be "its" in 'serializers.py' * Update 0004_geocode_fields.py correct use of "it's" to be "its" in '0004_geocode_fields.py', also fix spelling of 'syncronized' * Update 0029_auto_20200401_1006.py correct use of "it's" to be "its" in '0029_auto_20200401_1006.py', also fix spelling of 'syncronized' * Update models.py correct use of "it's" to be "its" in 'models.py', also fix spelling of 'syncronized' * Update email_confirm.html correct 're-initate' in email_confirm.html * Update notify-pdb-admin-user-affil.txt correct 'organzation' in 'notify-pdb-admin-user-affil.txt' * Update op_create.md correct 'organzation' in 'ip_create.md' * Update notify-pdb-admin-asnauto-skipvq.txt correct 'organzation' in 'notify-pdb-admin-asnauto-skipvq.txt' * Update models.py correct 'organzation' in 'models.py' * Update notify-pdb-admin-asnauto-skipvq.txt fix 'succesfully' in 'notify-pdb-admin-asnauto-skipvq.txt' * Update views.py fix 'succesfully' in 'views.py' * Update pdb_migrate_ixlans.py * Update models.py several more replacements of "it's" with "its" * Update rest.py one more replacement of "it's" with "its" * Update op_retrieve.md replace "dont" with "don't" * Update notify-org-admin-merge.txt fix "dont" with "don't" * Update error.html fix "dont" with "don't" * Update op_list.md fix "dont" with "don't" * Update serializers.py fix "dont" with "don't" * Update deskpro.py fix "dont" with "don't" * Update mock.py fix "dont" with "don't" * Update ixf.py fix "dont" with "don't" * Update admin.py fix "dont" with "don't" * Update api_schema.py fix "dont" with "don't" * Update signals.py fix "dont" with "don't" * Update views.py fix "dont" with "don't" * Update test_admin.py fix "dont" with "don't" * Update pdb_api_test.py fix "dont" with "don't" * Update models.py fix "dont" with "don't" * revert to "IXP" in help text Co-authored-by: Theo Baschak <tbaschak@users.noreply.github.com> Co-authored-by: Stefan Pratter <stefan@20c.com>
539 lines
15 KiB
Python
539 lines
15 KiB
Python
"""
|
|
Views for organization administrative actions
|
|
"""
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.views.decorators.csrf import csrf_protect
|
|
from django.http import JsonResponse
|
|
from django.template import loader
|
|
from django.conf import settings
|
|
from .forms import OrgAdminUserPermissionForm
|
|
|
|
from peeringdb_server.models import (
|
|
User,
|
|
Organization,
|
|
Network,
|
|
NetworkContact,
|
|
InternetExchange,
|
|
Facility,
|
|
UserOrgAffiliationRequest,
|
|
)
|
|
|
|
import django_namespace_perms.util as nsp
|
|
from django_namespace_perms.constants import *
|
|
from django_namespace_perms.models import UserPermission
|
|
|
|
from django_handleref.models import HandleRefModel
|
|
|
|
from django.utils.translation import ugettext_lazy as _
|
|
from django.utils.translation import override
|
|
|
|
|
|
def save_user_permissions(org, user, perms):
|
|
"""
|
|
Save user permissions for the specified org and user
|
|
|
|
perms should be a dict of permissioning ids and permission levels
|
|
"""
|
|
|
|
# wipe all the user's perms for the targeted org
|
|
|
|
user.userpermission_set.filter(namespace__startswith=org.nsp_namespace).delete()
|
|
|
|
# collect permissioning namespaces from the provided permissioning ids
|
|
|
|
nsp_perms = {}
|
|
|
|
for id, permissions in list(perms.items()):
|
|
|
|
if not permissions & PERM_READ:
|
|
permissions = permissions | PERM_READ
|
|
|
|
if id == "org.%d" % org.id:
|
|
nsp_perms[org.nsp_namespace] = permissions
|
|
nsp_perms[
|
|
NetworkContact.nsp_namespace_from_id(org.id, "*", "private")
|
|
] = permissions
|
|
elif id == "net":
|
|
nsp_perms[
|
|
Network.nsp_namespace_from_id(org.id, "*").strip(".*")
|
|
] = permissions
|
|
nsp_perms[
|
|
NetworkContact.nsp_namespace_from_id(org.id, "*", "private")
|
|
] = permissions
|
|
elif id == "ix":
|
|
nsp_perms[
|
|
InternetExchange.nsp_namespace_from_id(org.id, "*").strip(".*")
|
|
] = permissions
|
|
elif id == "fac":
|
|
nsp_perms[
|
|
Facility.nsp_namespace_from_id(org.id, "*").strip(".*")
|
|
] = permissions
|
|
elif id.find(".") > -1:
|
|
id = id.split(".")
|
|
if id[0] == "net":
|
|
nsp_perms[Network.nsp_namespace_from_id(org.id, id[1])] = permissions
|
|
nsp_perms[
|
|
NetworkContact.nsp_namespace_from_id(org.id, id[1], "private")
|
|
] = permissions
|
|
elif id[0] == "ix":
|
|
nsp_perms[
|
|
InternetExchange.nsp_namespace_from_id(org.id, id[1])
|
|
] = permissions
|
|
elif id[0] == "fac":
|
|
nsp_perms[Facility.nsp_namespace_from_id(org.id, id[1])] = permissions
|
|
|
|
# save
|
|
for ns, p in list(nsp_perms.items()):
|
|
UserPermission.objects.create(namespace=ns, permissions=p, user=user)
|
|
|
|
return nsp_perms
|
|
|
|
|
|
def load_all_user_permissions(org):
|
|
"""
|
|
Returns dict of all users with all their permissions for
|
|
the given org
|
|
"""
|
|
|
|
rv = {}
|
|
for user in org.usergroup.user_set.all():
|
|
uperms, perms = load_user_permissions(org, user)
|
|
rv[user.id] = {
|
|
"id": user.id,
|
|
"perms": perms,
|
|
"name": f"{user.full_name} <{user.email}> {user.username}",
|
|
}
|
|
return rv
|
|
|
|
|
|
def load_user_permissions(org, user):
|
|
"""
|
|
Returns user's permissions for the specified org
|
|
"""
|
|
|
|
# load all of the user's permissions related to this org
|
|
uperms = {
|
|
p.namespace: p.permissions
|
|
for p in user.userpermission_set.filter(namespace__startswith=org.nsp_namespace)
|
|
}
|
|
|
|
perms = {}
|
|
|
|
extract_permission_id(uperms, perms, org, org)
|
|
|
|
# extract user's permissioning ids from nsp_namespaces targeting
|
|
# organization's entities
|
|
for model in [Network, InternetExchange, Facility]:
|
|
extract_permission_id(uperms, perms, model, org)
|
|
|
|
# extract user's permissioning ids from nsp_namespaces targeting
|
|
# organization's entities by their id (eg user has perms only
|
|
# to THAT specific network)
|
|
for net in org.net_set_active:
|
|
extract_permission_id(uperms, perms, net, org)
|
|
|
|
for net in org.ix_set_active:
|
|
extract_permission_id(uperms, perms, net, org)
|
|
|
|
for net in org.fac_set_active:
|
|
extract_permission_id(uperms, perms, net, org)
|
|
|
|
return uperms, perms
|
|
|
|
|
|
def permission_ids(org):
|
|
"""
|
|
returns a dict of a valid permissioning ids for
|
|
the specified organization
|
|
"""
|
|
|
|
perms = {
|
|
"org.%d" % org.id: _("Organization and all Entities it owns"),
|
|
"net": _("Any Network"),
|
|
"fac": _("Any Facility"),
|
|
"ix": _("Any Exchange"),
|
|
}
|
|
|
|
perms.update(
|
|
{
|
|
"net.%d" % net.id: _("Network - %(net_name)s") % {"net_name": net.name}
|
|
for net in org.net_set_active
|
|
}
|
|
)
|
|
|
|
perms.update(
|
|
{
|
|
"ix.%d" % ix.id: _("Exchange - %(ix_name)s") % {"ix_name": ix.name}
|
|
for ix in org.ix_set_active
|
|
}
|
|
)
|
|
|
|
perms.update(
|
|
{
|
|
"fac.%d" % fac.id: _("Facility - %(fac_name)s") % {"fac_name": fac.name}
|
|
for fac in org.fac_set_active
|
|
}
|
|
)
|
|
|
|
return perms
|
|
|
|
|
|
def extract_permission_id(source, dest, entity, org):
|
|
"""
|
|
extract a user's permissioning id for the specified
|
|
entity from source <dict> and store it in dest <dict>
|
|
|
|
source should be a dict containing django-namespace-perms
|
|
(namespace, level) items
|
|
|
|
dest should be a dict where permission ids are to be
|
|
exracted to
|
|
|
|
entity can either be a HandleRef instance or clas
|
|
|
|
org needs to be an Organization instance that owns the
|
|
entity
|
|
"""
|
|
|
|
if isinstance(entity, HandleRefModel):
|
|
# instance
|
|
k = entity.nsp_namespace
|
|
j = "%s.%d" % (entity.ref_tag, entity.id)
|
|
else:
|
|
# class
|
|
k = entity.nsp_namespace_from_id(org.id, "*").strip(".*")
|
|
j = entity.handleref.tag
|
|
if k in source:
|
|
dest[j] = source[k]
|
|
|
|
|
|
def org_admin_required(fnc):
|
|
"""
|
|
Decorator function that ensures that the requesting user
|
|
has administrative rights to the targeted organization
|
|
|
|
Also sets "org" in kwargs
|
|
"""
|
|
|
|
def callback(request, **kwargs):
|
|
org_id = request.POST.get("org_id", request.GET.get("org_id"))
|
|
|
|
if not org_id:
|
|
return JsonResponse({}, status=400)
|
|
|
|
try:
|
|
org = Organization.objects.get(id=org_id)
|
|
if not nsp.has_perms(request.user, org.nsp_namespace_manage, "update"):
|
|
return JsonResponse({}, status=403)
|
|
kwargs["org"] = org
|
|
return fnc(request, **kwargs)
|
|
except Organization.DoesNotExist:
|
|
return JsonResponse(
|
|
{"non_field_errors": [_("Invalid organization specified")]}, status=400
|
|
)
|
|
|
|
return callback
|
|
|
|
|
|
def target_user_validate(fnc):
|
|
"""
|
|
Decorator function that ensures that the targeted user
|
|
is a member of the targeted organization
|
|
|
|
Should be below org_admin_required
|
|
|
|
Also sets "user" in kwargs
|
|
"""
|
|
|
|
def callback(request, **kwargs):
|
|
|
|
user_id = request.POST.get("user_id", request.GET.get("user_id"))
|
|
org = kwargs.get("org")
|
|
|
|
if not user_id:
|
|
return JsonResponse({}, status=400)
|
|
|
|
try:
|
|
user = User.objects.get(id=user_id)
|
|
except User.DoesNotExist:
|
|
return JsonResponse({}, status=400)
|
|
|
|
if not user.is_org_member(org) and not user.is_org_admin(org):
|
|
return JsonResponse({}, status=403)
|
|
|
|
kwargs["user"] = user
|
|
return fnc(request, **kwargs)
|
|
|
|
return callback
|
|
|
|
|
|
@login_required
|
|
@org_admin_required
|
|
def users(request, **kwargs):
|
|
"""
|
|
Returns JsonResponse with a list of all users in the specified org
|
|
"""
|
|
|
|
org = kwargs.get("org")
|
|
|
|
rv = {
|
|
"users": [
|
|
{
|
|
"id": user.id,
|
|
"name": f"{user.full_name} <{user.email}, {user.username}>",
|
|
}
|
|
for user in org.usergroup.user_set.all()
|
|
]
|
|
}
|
|
|
|
rv.update({"status": "ok"})
|
|
|
|
return JsonResponse(rv)
|
|
|
|
|
|
@login_required
|
|
@org_admin_required
|
|
@target_user_validate
|
|
def manage_user_delete(request, **kwargs):
|
|
"""
|
|
remove user from org
|
|
"""
|
|
|
|
org = kwargs.get("org")
|
|
user = kwargs.get("user")
|
|
|
|
save_user_permissions(org, user, {})
|
|
org.usergroup.user_set.remove(user)
|
|
org.admin_usergroup.user_set.remove(user)
|
|
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
|
|
@login_required
|
|
@org_admin_required
|
|
@target_user_validate
|
|
def manage_user_update(request, **kwargs):
|
|
"""
|
|
udpate a user in the org
|
|
|
|
right now this only allows for moving the user either
|
|
to admin or member group
|
|
"""
|
|
|
|
org = kwargs.get("org")
|
|
user = kwargs.get("user")
|
|
group = request.POST.get("group")
|
|
if group not in ["member", "admin"]:
|
|
return JsonResponse({"group": _("Needs to be member or admin")}, status=400)
|
|
|
|
if group == "admin":
|
|
org.usergroup.user_set.remove(user)
|
|
org.admin_usergroup.user_set.add(user)
|
|
elif group == "member":
|
|
org.usergroup.user_set.add(user)
|
|
org.admin_usergroup.user_set.remove(user)
|
|
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
|
|
@login_required
|
|
@org_admin_required
|
|
def user_permissions(request, **kwargs):
|
|
"""
|
|
Returns JsonRespone with list of user's permissions for the targeted
|
|
org an entities under it
|
|
|
|
Permisions are returned as a dict of permissioning ids and permission
|
|
levels.
|
|
|
|
Permissioning ids serve as a wrapper for actual permissioning namespaces
|
|
so we can expose them to the organization admins for changes without allowing
|
|
them to set permissioning namespaces directly.
|
|
"""
|
|
|
|
org = kwargs.get("org")
|
|
perms_rv = {}
|
|
for user in org.usergroup.user_set.all():
|
|
uperms, perms = load_user_permissions(org, user)
|
|
perms_rv[user.id] = perms
|
|
|
|
return JsonResponse({"status": "ok", "user_permissions": perms_rv})
|
|
|
|
|
|
@login_required
|
|
@csrf_protect
|
|
@org_admin_required
|
|
@target_user_validate
|
|
def user_permission_update(request, **kwargs):
|
|
"""
|
|
Update/Add a user's permission
|
|
|
|
perms = permission level
|
|
entity = permission id
|
|
"""
|
|
|
|
org = kwargs.get("org")
|
|
user = kwargs.get("user")
|
|
|
|
uperms, perms = load_user_permissions(org, user)
|
|
form = OrgAdminUserPermissionForm(request.POST)
|
|
if not form.is_valid():
|
|
return JsonResponse(form.errors, status=400)
|
|
|
|
level = form.cleaned_data.get("perms")
|
|
entity = form.cleaned_data.get("entity")
|
|
perms[entity] = level
|
|
save_user_permissions(org, user, perms)
|
|
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
|
|
@login_required
|
|
@csrf_protect
|
|
@org_admin_required
|
|
@target_user_validate
|
|
def user_permission_remove(request, **kwargs):
|
|
"""
|
|
Remove a user's permission
|
|
|
|
entity = permission id
|
|
"""
|
|
|
|
org = kwargs.get("org")
|
|
user = kwargs.get("user")
|
|
entity = request.POST.get("entity")
|
|
uperms, perms = load_user_permissions(org, user)
|
|
if entity in perms:
|
|
del perms[entity]
|
|
save_user_permissions(org, user, perms)
|
|
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
|
|
@login_required
|
|
@org_admin_required
|
|
def permissions(request, **kwargs):
|
|
"""
|
|
Returns list of permissioning ids with labels that
|
|
are valid to be permissioned out to regular org users
|
|
|
|
Permissioning ids serve as a wrapper for actual permissioning namespaces
|
|
so we can expose them to the organization admins for changes without allowing
|
|
them to set permissioning namespaces directly.
|
|
"""
|
|
|
|
org = kwargs.get("org")
|
|
|
|
perms = [{"id": id, "name": name} for id, name in list(permission_ids(org).items())]
|
|
perms = sorted(perms, key=lambda x: x.get("name"))
|
|
return JsonResponse({"status": "ok", "permissions": perms})
|
|
|
|
|
|
@login_required
|
|
@csrf_protect
|
|
@org_admin_required
|
|
def uoar_approve(request, **kwargs):
|
|
"""
|
|
Approve a user request to affiliate with the organization
|
|
"""
|
|
|
|
org = kwargs.get("org")
|
|
|
|
try:
|
|
uoar = UserOrgAffiliationRequest.objects.get(id=request.POST.get("id"))
|
|
if uoar.org != org:
|
|
return JsonResponse({}, status=403)
|
|
|
|
try:
|
|
user = uoar.user
|
|
except User.DoesNotExist:
|
|
uoar.delete()
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
uoar.approve()
|
|
|
|
# notify rest of org admins that the affiliation request has been
|
|
# approved
|
|
|
|
for admin_user in org.admin_usergroup.user_set.all():
|
|
if admin_user != request.user:
|
|
with override(admin_user.locale):
|
|
admin_user.email_user(
|
|
_("%(user_name)s's affiliation request has been approved")
|
|
% {"user_name": uoar.user.full_name},
|
|
loader.get_template(
|
|
"email/notify-org-admin-user-affil-approved.txt"
|
|
).render(
|
|
{
|
|
"user": request.user,
|
|
"uoar": uoar,
|
|
"org_management_url": "%s/org/%d#users"
|
|
% (settings.BASE_URL, org.id),
|
|
}
|
|
),
|
|
)
|
|
|
|
return JsonResponse(
|
|
{
|
|
"status": "ok",
|
|
"full_name": user.full_name,
|
|
"id": user.id,
|
|
"email": user.email,
|
|
}
|
|
)
|
|
|
|
except UserOrgAffiliationRequest.DoesNotExist:
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
|
|
@login_required
|
|
@csrf_protect
|
|
@org_admin_required
|
|
def uoar_deny(request, **kwargs):
|
|
"""
|
|
Approve a user request to affiliate with the organization
|
|
"""
|
|
|
|
org = kwargs.get("org")
|
|
|
|
try:
|
|
uoar = UserOrgAffiliationRequest.objects.get(id=request.POST.get("id"))
|
|
if uoar.org != org:
|
|
return JsonResponse({}, status=403)
|
|
|
|
try:
|
|
user = uoar.user
|
|
uoar.deny()
|
|
|
|
except User.DoesNotExist:
|
|
uoar.delete()
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
# notify rest of org admins that the affiliation request has been
|
|
# approved
|
|
|
|
for user in org.admin_usergroup.user_set.all():
|
|
if user != request.user:
|
|
with override(user.locale):
|
|
user.email_user(
|
|
_("%(user_name)s's affiliation request has been denied")
|
|
% {"user_name": uoar.user.full_name},
|
|
loader.get_template(
|
|
"email/notify-org-admin-user-affil-denied.txt"
|
|
).render(
|
|
{
|
|
"user": request.user,
|
|
"uoar": uoar,
|
|
"org_management_url": "%s/org/%d#users"
|
|
% (settings.BASE_URL, org.id),
|
|
}
|
|
),
|
|
)
|
|
|
|
except UserOrgAffiliationRequest.DoesNotExist:
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
return JsonResponse({"status": "ok"})
|