mirror/peeringdb-peeringdb
1
0
Fork 0
mirror of https://github.com/peeringdb/peeringdb.git synced 2024-05-11 05:55:09 +00:00
Code Releases Activity

Merge branch 'support_202306' into 'support_202306_base'

Browse Source 
Support 202306

See merge request gh/peeringdb/peeringdb!393
This commit is contained in:
Stefan Pratter
2023-07-11 11:55:12 +00:00
parent 95767908a6 e6eda5e5a5
commit 63b86fcea4
47 changed files with 1794 additions and 504 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Ctl/dev/auto_pdb_load_data.sh
Executable
+40
View File
@@ -0,0 +1,40 @@
#!/bin/bash
AUTO_PDB_LOAD_DATA_DIR="/srv/www.peeringdb.com/peeringdb_server/scripts/auto_pdb_load_data.py"
PID=$(docker exec peeringdb pgrep -f "python $AUTO_PDB_LOAD_DATA_DIR")
if [ "$1" = "" ]; then
echo "use '-h' or 'help' for usage"
exit 1
elif [ "$1" = "start" ]; then
if [[ -n $PID ]]; then
echo "The automated pdb_load_data already started"
exit 1
else
COMMAND="python $AUTO_PDB_LOAD_DATA_DIR >> /var/log/auto_pdb_load_data.log 2>&1 &"
echo "Starting the automated pdb_load_data"
fi
elif [ "$1" = "stop" ]; then
if [[ -n $PID ]]; then
COMMAND="kill $PID"
echo "The automated pdb_load_data has been stopped"
else
echo "cron process not found"
exit 1
fi
elif [ "$1" = "log" ]; then
COMMAND="cat /var/log/auto_pdb_load_data.log"
elif [ "$1" = "-h" ] || [ "$1" = "help" ]; then
echo "Usage: ./Ctl/dev/auto_pdb_load_data.sh"
echo ""
echo "Available options:"
echo " start Start automated service to run pdb_load_data"
echo " stop Stop automated pdb_load_data command"
echo " log Show log"
exit 1
else
echo "$1 is not an option, use '-h' or 'help' for usage"
exit 1
fi
docker exec peeringdb sh -c "$COMMAND"
Ctl/dev/docker-compose.yml
+2
View File
@@ -22,9 +22,11 @@ services:
context: ../..
dockerfile: Dockerfile
target: tester
container_name: peeringdb
command: runserver 0.0.0.0:8000
env_file: .env
depends_on:
- database
- elasticsearch
environment:
DATABASE_USER: peeringdb
Ctl/dev/setup.sh
Executable
+32
View File
@@ -0,0 +1,32 @@
#!/bin/bash
if [ "$1" = "" ]; then
cron="on"
elif [ "$1" = "disable" ]; then
cron="off"
elif [ "$1" = "-h" ] || [ "$1" = "help" ]; then
echo "Usage: ./Ctl/dev/setup.sh"
echo ""
echo "Available options:"
echo " disable Disable automated generate data from pdb_load_data command"
echo " help Show available commands"
exit 1
else
echo "$1 is not an option, use '-h' or 'help' for usage"
exit 1
fi
./Ctl/dev/compose.sh build peeringdb
./Ctl/dev/compose.sh up -d database
./Ctl/dev/run.sh migrate
until ./Ctl/dev/run.sh migrate; do
echo "Migrations failed. Retrying in 5 seconds..."
sleep 5
done
./Ctl/dev/run.sh loaddata fixtures/initial_data.json
./Ctl/dev/run.sh createsuperuser
./Ctl/dev/run.sh createcachetable
./Ctl/dev/compose.sh up -d peeringdb
if [ "$cron" = "on" ]; then
./Ctl/dev/auto_pdb_load_data.sh start
fi
Ctl/docker/entrypoint.sh
+2
View File
@@ -12,6 +12,7 @@ function migrate() {
cd /srv/www.peeringdb.com
case "$1" in
"uwsgi" )
echo starting uwsgi
@@ -35,6 +36,7 @@ case "$1" in
export DATABASE_USER=root
export DATABASE_PASSWORD=""
export RELEASE_ENV=run_tests
export PEERINGDB_SYNC_CACHE_URL=""
unset BASE_URL
unset OAUTH2_PROVIDER_APPLICATION_MODEL
unset SESSION_COOKIE_DOMAIN
Ctl/local/auto_pdb_load_data.sh
Executable
+40
View File
@@ -0,0 +1,40 @@
#!/bin/bash
AUTO_PDB_LOAD_DATA_DIR="/srv/www.peeringdb.com/peeringdb_server/scripts/auto_pdb_load_data.py"
PID=$(docker exec peeringdb pgrep -f "python $AUTO_PDB_LOAD_DATA_DIR")
if [ "$1" = "" ]; then
echo "use '-h' or 'help' for usage"
exit 1
elif [ "$1" = "start" ]; then
if [[ -n $PID ]]; then
echo "The automated pdb_load_data already started"
exit 1
else
COMMAND="python $AUTO_PDB_LOAD_DATA_DIR >> /var/log/auto_pdb_load_data.log 2>&1 &"
echo "Starting the automated pdb_load_data"
fi
elif [ "$1" = "stop" ]; then
if [[ -n $PID ]]; then
COMMAND="kill $PID"
echo "The automated pdb_load_data has been stopped"
else
echo "cron process not found"
exit 1
fi
elif [ "$1" = "log" ]; then
COMMAND="cat /var/log/auto_pdb_load_data.log"
elif [ "$1" = "-h" ] || [ "$1" = "help" ]; then
echo "Usage: ./Ctl/local/auto_pdb_load_data.sh"
echo ""
echo "Available options:"
echo " start Start automated service to run pdb_load_data"
echo " stop Stop automated pdb_load_data command"
echo " log Show log"
exit 1
else
echo "$1 is not an option, use '-h' or 'help' for usage"
exit 1
fi
docker exec peeringdb sh -c "$COMMAND"
Ctl/local/compose.sh
Executable
+10
View File
@@ -0,0 +1,10 @@
#!/bin/bash
COMPOSE_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
PROJECT_NAME=$(basename $(git rev-parse --show-toplevel))$(basename $COMPOSE_DIR)
docker-compose -p $PROJECT_NAME -f $COMPOSE_DIR/docker-compose.yml $@
if [[ "$@" == *"up -d peeringdb"* ]]; then
./Ctl/local/auto_pdb_load_data.sh start
fi
Ctl/local/docker-compose.yml
+49
View File
@@ -0,0 +1,49 @@
version: "3.4"
services:
database:
image: "mysql:8"
environment:
MYSQL_DATABASE: peeringdb
MYSQL_USER: peeringdb
MYSQL_PASSWORD: devPASSWORD
MYSQL_ALLOW_EMPTY_PASSWORD: 1
restart: always
ports:
- "23306:3306"
volumes:
- peeringdb_database:/var/lib/mysql:Z
networks:
- peeringdb_local_network
peeringdb:
user: "0:0"
image: ghcr.io/peeringdb/peeringdb-server:latest
container_name: peeringdb
command: runserver 0.0.0.0:8000
env_file: .env
depends_on:
- database
- elasticsearch
environment:
DATABASE_USER: peeringdb
DATABASE_PASSWORD: devPASSWORD
DATABASE_HOST: database
ports:
- "${DJANGO_PORT:-8000}:8000"
networks:
- peeringdb_local_network
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:7.6.1
ports:
- "19200:9200"
environment:
- discovery.type=single-node
networks:
- peeringdb_local_network
volumes:
peeringdb_database:
networks:
peeringdb_local_network:
Ctl/local/env.example
View File
Ctl/local/exec.sh
Executable
+6
View File
@@ -0,0 +1,6 @@
#!/bin/bash
COMPOSE_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
PROJECT_NAME=$(basename $(git rev-parse --show-toplevel))$(basename $COMPOSE_DIR)
docker-compose -p $PROJECT_NAME -f $COMPOSE_DIR/docker-compose.yml exec peeringdb manage $@
Ctl/local/run.sh
Executable
+6
View File
@@ -0,0 +1,6 @@
#!/bin/bash
COMPOSE_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
PROJECT_NAME=$(basename $(git rev-parse --show-toplevel))$(basename $COMPOSE_DIR)
docker-compose -p $PROJECT_NAME -f $COMPOSE_DIR/docker-compose.yml run $run_options --rm peeringdb $@
Ctl/local/setup.sh
Executable
+58
View File
@@ -0,0 +1,58 @@
#!/bin/bash
if [ "$1" = "" ]; then
cron="on"
elif [ "$1" = "disable" ]; then
cron="off"
elif [ "$1" = "-h" ] || [ "$1" = "help" ]; then
echo "Usage: ./Ctl/local/setup.sh"
echo ""
echo "Available options:"
echo " disable Disable automated generate data from pdb_load_data command"
echo " help Show available commands"
exit 1
else
echo "$1 is not an option, use '-h' or 'help' for usage"
exit 1
fi
# Check if .env file exists, if not copy from example.env
if [ ! -f ./Ctl/local/.env ]; then
cp ./Ctl/local/env.example ./Ctl/local/.env
fi
./Ctl/local/compose.sh build peeringdb
./Ctl/local/compose.sh up -d database
./Ctl/local/run.sh migrate
until ./Ctl/local/run.sh migrate; do
echo "Migrations failed. Retrying in 5 seconds..."
sleep 5
done
./Ctl/local/run.sh loaddata fixtures/initial_data.json
#echo "--- Creating superuser ---"
#echo ""
#./Ctl/local/run.sh createsuperuser
./Ctl/local/run.sh createcachetable
echo "--- Starting PeeringDB ---"
echo ""
./Ctl/local/compose.sh up -d peeringdb
echo "--- Loading PeeringDB data (this can take a while) ---"
echo ""
./Ctl/local/run.sh pdb_load_data --commit
echo "--- Updating search index ---"
echo ""
./Ctl/local/run.sh search_index -f --rebuild
echo "--- Fetching api-cache ---"
echo ""
./Ctl/local/exec.sh pdb_fetch_api_cache
if [ "$cron" = "on" ]; then
echo "Starting automatic sync daemon"
./Ctl/local/auto_pdb_load_data.sh start
fi
docs/local_setup.md
+58
View File
@@ -0,0 +1,58 @@
# Setting up a local peeringdb instance
Note: if you're looking at setting up a developer instance please refer to the development instructions instead.
This will set up an instance you can use to run a local snapshot of peeringdb.
Usage:
```sh
./Ctl/local/setup.sh [option]
```
This is a bash script used to set up a local PeeringDB environment. The script takes an optional argument which can be either "disable" or "help". If no argument is provided, the script assumes that automated data synchronization from the production API is enabled.
The script checks if the .env file exists, if not, it copies from env.example.
It then builds the peeringdb container, sets up the database, runs migrations, loads initial data, creates a cache table, starts the PeeringDB container, loads PeeringDB data, updates the search index, fetches api-cache and if automated data generation is enabled, it starts the automatic sync daemon.
Options:
- disable: Disable automated generate data from pdb_load_data command
- help: Show available commands
# Manually interacting with the automatic data sync
This is a bash script used to manage the automated pdb_load_data service. The script takes one required argument which can be either "start", "stop", "log" or "help".
The "start" option starts the automated pdb_load_data service if it's not already running. The "stop" option stops the automated pdb_load_data service if it's running. The "log" option shows the log of the automated pdb_load_data service. The "help" option shows the available commands.
Usage:
```sh
./Ctl/local/auto_pdb_load_data.sh [option]
```
Options:
- start: Start automated service to run pdb_load_data
- stop: Stop automated pdb_load_data command
- log: Show log
- help: Show available commands
# Example: automatic api cache regeneration
Currently automatic api cache updates aren't scheduled on local instances, as they can take up a lot of resources. Here is an example of a python script that will regenerate the API cache of your instance every 1000 seconds.
```py
import subprocess
import time
def run_commands():
"""
Run the specified shell commands every 5 minutes.
"""
while True:
subprocess.run(["Ctl/local/exec.sh", "pdb_api_cache", "--depth", "0"])
time.sleep(1000)
run_commands()
```
mainsite/oauth2/validators.py
+4
View File
@@ -5,6 +5,10 @@ from mainsite.oauth2.scopes import SupportedScopes
class OIDCValidator(OAuth2Validator):
# https://django-oauth-toolkit.readthedocs.io/en/latest/changelog.html#id12
oidc_claim_scope = None
def get_additional_claims(self):
"""PeeringDB-specific claims added to the standard claims defined in a JWT token.
mainsite/settings/__init__.py
+14 -1
View File
@@ -811,6 +811,11 @@ CORS_ALLOW_METHODS = ["GET", "OPTIONS"]
# allows PeeringDB to use external OAuth2 sources
set_bool("OAUTH_ENABLED", False)
# https://django-oauth-toolkit.readthedocs.io/en/latest/changelog.html#id12
# default changed to True in django-oauth-toolkit v2, set to False for now
# to avoid breaking existing clients
set_bool("PKCE_REQUIRED", False)
# enables OpenID Connect support
set_bool("OIDC_ENABLED", True)
@@ -840,6 +845,7 @@ MIDDLEWARE += (
OAUTH2_PROVIDER = {
"OIDC_ENABLED": OIDC_ENABLED,
"OIDC_RSA_PRIVATE_KEY": OIDC_RSA_PRIVATE_KEY,
"PKCE_REQUIRED": PKCE_REQUIRED,
"OAUTH2_VALIDATOR_CLASS": "mainsite.oauth2.validators.OIDCValidator",
"SCOPES": {
SupportedScopes.OPENID: "OpenID Connect scope",
@@ -1234,10 +1240,13 @@ set_option("DELETE_ORPHANED_USER_DAYS", 90)
# Notify orphaned users n days before deletion
set_option("NOTIFY_ORPHANED_USER_DAYS", 30)
# Grace peruid before a newly created user can be flagged for deletion
# Grace period before a newly created user can be flagged for deletion
# This is so users have some time to affiliate naturally. (days)
set_option("MIN_AGE_ORPHANED_USER_DAYS", 14)
# Notification period to notify organizations of users missing 2FA (days)
set_option("NOTIFY_MISSING_2FA_DAYS", 30)
# pdb_validate_data cache timeout default
set_option("PDB_VALIDATE_DATA_CACHE_TIMEOUT", 3600)
@@ -1307,4 +1316,8 @@ set_option("PEERINGDB_SYNC_PASSWORD", "")
# If the api key is specified it will be used over the username and password
set_option("PEERINGDB_SYNC_API_KEY", "")
# peeringdb sync cache
set_option("PEERINGDB_SYNC_CACHE_URL", "https://cache.peeringdb.com/api")
set_option("PEERINGDB_SYNC_CACHE_DIR", os.path.join(BASE_DIR, "sync-cache"))
print_debug(f"loaded settings for PeeringDB {PEERINGDB_VERSION} (DEBUG: {DEBUG})")
peeringdb_server/admin.py
+8
View File
@@ -1713,6 +1713,14 @@ class UserOrgAffiliationRequestAdmin(ModelAdminWithUrlActions, ProtectedDeleteAd
request, _("Cannot approve a canceled affiliation request")
)
continue
if each.org.require_2fa and not each.user.has_2fa:
messages.error(
request,
_(
"Cannot approve while User has 2FA disabled - organization requires 2FA"
),
)
continue
each.approve()
each.notify_ownership_approved()
peeringdb_server/client_adaptor/backend.py
+8
View File
@@ -135,9 +135,17 @@ class Backend(BaseBackend):
setattr(obj, field.name, value)
if obj.HandleRef.tag == "ix":
obj._meta.get_field("updated").auto_now = False
obj._meta.get_field("created").auto_now = False
obj.save(create_ixlan=False)
obj._meta.get_field("updated").auto_now = True
obj._meta.get_field("created").auto_now = True
else:
obj._meta.get_field("updated").auto_now = False
obj._meta.get_field("created").auto_now = False
obj.save()
obj._meta.get_field("updated").auto_now = True
obj._meta.get_field("created").auto_now = True
def detect_uniqueness_error(self, exc):
"""
peeringdb_server/forms.py
+1
View File
@@ -250,6 +250,7 @@ class OrgUserOptions(forms.ModelForm):
class Meta:
model = Organization
fields = [
"require_2fa",
"restrict_user_emails",
"email_domains",
"periodic_reauth",
peeringdb_server/management/commands/pdb_api_test.py
+2 -93
View File
@@ -1714,6 +1714,8 @@ class TestJSON(unittest.TestCase):
"""
Tests the carrier-facility creation and approval
processes
As of #1344 all carrierfac objects are automatically approved.
"""
# create carrer-facility
@@ -1737,101 +1739,8 @@ class TestJSON(unittest.TestCase):
assert carrierfac.facility == facility
assert carrierfac.carrier == carrier
assert carrierfac.status == "pending"
# only the facility org can approve
response = self.db_org_admin._request(
f"carrierfac/{carrierfac.id}/approve", method="POST", data="{}"
)
assert response.status_code == 403
# give permissions to approve the carrier facility (same as facility org)
if hasattr(self, "org_key"):
self.org_key.grainy_permissions.add_permission(
f"peeringdb.organization.{facility.org.id}.facility", 15
)
else:
self.db_org_admin.user_inst.grainy_permissions.add_permission(
f"peeringdb.organization.{facility.org.id}.facility", 15
)
# need to re-init rest client so permissions stick
self.db_org_admin = self.rest_client(URL, verbose=VERBOSE, **USER_ORG_ADMIN)
response = self.db_org_admin._request(
f"carrierfac/{carrierfac.id}/approve", method="POST", data="{}"
)
assert response.status_code == 200
carrierfac.refresh_from_db()
assert carrierfac.status == "ok"
##########################################################################
def test_org_admin_002_reject_carrierfac(self):
"""
Tests the carrier-facility creation and rejection
processes
"""
# create carrier-facility
carrier = SHARED["carrier_rw_ok"]
facility = SHARED["fac_r_ok"]
data = {"carrier_id": carrier.id, "fac_id": facility.id}
r_data = (
self.db_org_admin._request("carrierfac", method="POST", data=data)
.json()
.get("data")[0]
)
SHARED["carrierfac_id"] = r_data.get("id")
carrierfac = CarrierFacility.objects.get(id=SHARED["carrierfac_id"])
# assert that the carrier-facility is status pending (needs to be rejected)
assert carrierfac.facility == facility
assert carrierfac.carrier == carrier
assert carrierfac.status == "pending"
# only the facility org can reject
response = self.db_org_admin._request(
f"carrierfac/{carrierfac.id}/reject", method="POST", data="{}"
)
assert response.status_code == 403
# give permissions to approve the carrier facility (same as facility org)
if hasattr(self, "org_key"):
self.org_key.grainy_permissions.add_permission(
f"peeringdb.organization.{facility.org.id}.facility", 15
)
else:
self.db_org_admin.user_inst.grainy_permissions.add_permission(
f"peeringdb.organization.{facility.org.id}.facility", 15
)
# need to re-init rest client so permissions stick
self.db_org_admin = self.rest_client(URL, verbose=VERBOSE, **USER_ORG_ADMIN)
response = self.db_org_admin._request(
f"carrierfac/{carrierfac.id}/reject", method="POST", data="{}"
)
assert response.status_code == 200
carrierfac.refresh_from_db()
assert carrierfac.status == "deleted"
def test_org_admin_002_auto_approve_carrierfac(self):
"""
Tests the carrier-facility creation and AUTO approval
peeringdb_server/management/commands/pdb_fetch_api_cache.py
+46
View File
@@ -0,0 +1,46 @@
"""
Django management command
Will fetch api cache files from PEERINGDB_SYNC_CACHE_URL to API_CACHE_ROOT
"""
import json
import os
import sys
from urllib.parse import urljoin
import requests
from django.conf import settings
from django.core.management.base import BaseCommand
from peeringdb_server.models import REFTAG_MAP
class Command(BaseCommand):
help = "Fetch api cache files from PEERINGDB_SYNC_CACHE_URL to API_CACHE_ROOT"
def handle(self, *args, **options):
"""
Fetch api cache files from PEERINGDB_SYNC_CACHE_URL to API_CACHE_ROOT
"""
tags = REFTAG_MAP.keys()
for tag in tags:
url = urljoin(f"{settings.PEERINGDB_SYNC_CACHE_URL}/", tag) + "-0.json"
response = requests.get(url)
# Make sure the request was successful
if response.status_code == 200:
data = response.json()
# store the json data to a file in data/cache directory
cache_file_path = os.path.join(settings.API_CACHE_ROOT, f"{tag}-0.json")
with open(cache_file_path, "w") as file:
json.dump(data, file, indent=4)
self.stdout.write(f"Successfully fetched data for {tag} from {url}")
else:
self.stdout.write(
f"Error fetching data for {tag} from {url}: HTTP {response.status_code}"
)
sys.exit(1)
peeringdb_server/management/commands/pdb_load_data.py
+19 -1
View File
@@ -63,6 +63,10 @@ class Command(BaseCommand):
"--commit", action="store_true", help="will commit the changes"
)
parser.add_argument(
"--no-cache", action="store_true", help="will not use the sync cache"
)
def handle(self, *args, **options):
if settings.RELEASE_ENV != "dev" and not settings.TUTORIAL_MODE:
self.stdout.write(
@@ -83,7 +87,14 @@ class Command(BaseCommand):
# settings.USE_TZ = True
db_settings = settings.DATABASES.get("default")
sync_options = {"url": options.get("url")}
sync_options = {
"url": options.get("url"),
"cache_url": settings.PEERINGDB_SYNC_CACHE_URL,
"cache_dir": settings.PEERINGDB_SYNC_CACHE_DIR,
}
if options.get("no_cache"):
sync_options["cache_url"] = ""
# allow authentication for the sync (set as env vars)
if settings.PEERINGDB_SYNC_API_KEY:
@@ -96,6 +107,8 @@ class Command(BaseCommand):
"No authentication configured for sync, using anonymous access - this may cause rate limiting issues. Set PEERINGDB_SYNC_API_KEY or PEERINGDB_SYNC_USERNAME and PEERINGDB_SYNC_PASSWORD to configure authentication."
)
print("Syncing data from {url}".format(**sync_options))
config = {
"sync": sync_options,
"orm": {
@@ -128,5 +141,10 @@ class Command(BaseCommand):
# disable validation of parent status during sync
settings.DATA_QUALITY_VALIDATE_PARENT_STATUS = False
# disable elasticsearch auto indexing if there are zero records
if pdb_models.Organization.objects.count() == 0:
settings.ELASTICSEARCH_DSL_AUTOSYNC = False
settings.ELASTICSEARCH_DSL_AUTO_REFRESH = False
client = Client(config)
client.update_all(resource.all_resources(), since=None)
peeringdb_server/management/commands/pdb_notify_2fa.py
+108
View File
@@ -0,0 +1,108 @@
"""
Load initial data from another peeringdb instance using the REST API.
"""
from datetime import timedelta
from django.conf import settings
from django.core.management.base import BaseCommand
from django.template import loader
from django.urls import reverse
from django.utils import timezone
from peeringdb_server.models import Organization
class Command(BaseCommand):
help = "Notify the organization user admins if the member of the organization that required 2fa has a member without 2fa enabled"
def add_arguments(self, parser):
parser.add_argument(
"--commit", action="store_true", help="will commit the changes"
)
parser.add_argument("--limit", type=int, default=10)
def handle(self, *args, **options):
threshold = timezone.now() - timedelta(days=settings.NOTIFY_MISSING_2FA_DAYS)
orgs_require_2fa = Organization.objects.filter(
require_2fa=True,
last_notified__lt=threshold,
status="ok",
) | Organization.objects.filter(
require_2fa=True,
last_notified=None,
status="ok",
)
if options.get("limit"):
limit = options.get("limit")
orgs_require_2fa = orgs_require_2fa[:limit]
if any(
not user.has_2fa
for org in Organization.objects.all()
for user in org.all_users
):
for org in orgs_require_2fa:
users_without_2fa = []
if org.all_users:
org_users_without_2fa = [
user for user in org.all_users if not user.has_2fa
]
users_without_2fa.extend(org_users_without_2fa)
if users_without_2fa:
if not options.get("commit"):
self.stdout.write(f"{org} would get emailed")
continue
self.notify_org(org, users_without_2fa)
if not options.get("commit"):
self.stdout.write(
"Run the command with `--commit` if you are sure you want "
"to send notification to the organization admins.".format(**options)
)
def notify_org(self, org, users):
users_without_2fa = []
for user in users:
users_without_2fa = []
action_url = f"{settings.BASE_URL}{reverse('handle-2fa')}?org={org.id}&member={user.id}"
message = loader.get_template(
"email/notify-org-member-disable-2fa.txt"
).render(
{
"org": org.name,
"member": user,
"drop_member_url": f"{action_url}&action=drop",
"leave_member_affiliate_url": f"{action_url}&action=leave",
"cancel_2fa_url": f"{action_url}&action=disable",
"org_url": f"{settings.BASE_URL}/org/{org.id}#users",
"support_email": settings.DEFAULT_FROM_EMAIL,
}
)
users_without_2fa.append(message)
for admin_user in org.admin_usergroup.user_set.all():
admin_user.email_user(
subject=f"The following users affiliated with {org.name} organization do not have 2FA turned on",
message=loader.get_template(
"email/notify-org-user-admin-2fa.txt"
).render(
{
"org_name": org.name,
"users_without_2fa": users_without_2fa,
"org_url": f"{settings.BASE_URL}/org/{org.id}#users",
"support_email": settings.DEFAULT_FROM_EMAIL,
}
),
)
org.last_notified = (
timezone.now()
) # Update last_notified field to the current time
org.save()
peeringdb_server/management/commands/pdb_wipe.py
+7
View File
@@ -43,6 +43,9 @@ class Command(BaseCommand):
self.load_data = options.get("load_data", False)
self.load_data_url = options.get("load_data_url")
settings.ELASTICSEARCH_DSL_AUTOSYNC = False
settings.ELASTICSEARCH_DSL_AUTO_REFRESH = False
if not settings.TUTORIAL_MODE:
self.log("Command can only be run with tutorial mode enabled")
return
@@ -75,6 +78,10 @@ class Command(BaseCommand):
self.log("Cleared seassions")
self.log(
"Search indexes will need to be manually updated (if applicable) using the search_index command"
)
if self.load_data:
call_command(
"pdb_load_data",
peeringdb_server/migrations/0111_update_carrierfac_status.py
+17
View File
@@ -0,0 +1,17 @@
from django.db import migrations
from peeringdb_server.models import CarrierFacility
def update_carrierfac_status(apps, schema_editor):
CarrierFacility.objects.filter(status="pending").update(status="ok")
class Migration(migrations.Migration):
dependencies = [
("peeringdb_server", "0110_social_media"),
]
operations = [
migrations.RunPython(update_carrierfac_status, migrations.RunPython.noop),
]
peeringdb_server/migrations/0112_organization_require_totpdevice_and_more.py
+65
View File
@@ -0,0 +1,65 @@
# Generated by Django 4.2 on 2023-06-24 04:31
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("peeringdb_server", "0111_update_carrierfac_status"),
]
operations = [
migrations.AddField(
model_name="organization",
name="require_2fa",
field=models.BooleanField(
default=False,
help_text="Require users in your organization to activate 2FA.",
),
),
migrations.AlterField(
model_name="oauthapplication",
name="user",
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name="%(app_label)s_%(class)s",
to=settings.AUTH_USER_MODEL,
),
),
migrations.AlterField(
model_name="organizationapikey",
name="hashed_key",
field=models.CharField(editable=False, max_length=150),
),
migrations.AlterField(
model_name="organizationapikey",
name="id",
field=models.CharField(
editable=False,
max_length=150,
primary_key=True,
serialize=False,
unique=True,
),
),
migrations.AlterField(
model_name="userapikey",
name="hashed_key",
field=models.CharField(editable=False, max_length=150),
),
migrations.AlterField(
model_name="userapikey",
name="id",
field=models.CharField(
editable=False,
max_length=150,
primary_key=True,
serialize=False,
unique=True,
),
),
]
peeringdb_server/migrations/0113_organization_last_notified.py
+22
View File
@@ -0,0 +1,22 @@
# Generated by Django 4.2.2 on 2023-07-03 09:18
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("peeringdb_server", "0112_organization_require_totpdevice_and_more"),
]
operations = [
migrations.AddField(
model_name="organization",
name="last_notified",
field=models.DateTimeField(
blank=True,
help_text="Date when the organization admins were notified about 2FA",
null=True,
),
),
]
peeringdb_server/migrations/0114_oauthapplication_post_logout_redirect_uris_and_more.py
+33
View File
@@ -0,0 +1,33 @@
# Generated by Django 4.2.3 on 2023-07-10 13:48
import oauth2_provider.generators
import oauth2_provider.models
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("peeringdb_server", "0113_organization_last_notified"),
]
operations = [
migrations.AddField(
model_name="oauthapplication",
name="post_logout_redirect_uris",
field=models.TextField(
blank=True, help_text="Allowed Post Logout URIs list, space separated"
),
),
migrations.AlterField(
model_name="oauthapplication",
name="client_secret",
field=oauth2_provider.models.ClientSecretField(
blank=True,
db_index=True,
default=oauth2_provider.generators.generate_client_secret,
help_text="Hashed on Save. Copy it now if this is a new secret.",
max_length=255,
),
),
]
peeringdb_server/migrations/0115_hash_oauth_client_secret.py
+29
View File
@@ -0,0 +1,29 @@
import oauth2_provider.generators
import oauth2_provider.models
from django.db import migrations
from oauth2_provider import settings
def forwards_func(apps, schema_editor):
"""
Forward migration touches every application.client_secret which will cause it to be hashed if not already the case.
"""
Application = apps.get_model(settings.APPLICATION_MODEL)
applications = Application._default_manager.all()
for application in applications:
application.save(update_fields=["client_secret"])
class Migration(migrations.Migration):
dependencies = [
("oauth2_provider", "0005_auto_20211222_2352"),
(
"peeringdb_server",
"0114_oauthapplication_post_logout_redirect_uris_and_more",
),
]
operations = [
migrations.RunPython(forwards_func),
]
peeringdb_server/mock.py
+3
View File
@@ -350,3 +350,6 @@ class Mock:
def email_domains(self, data, reftag=None):
return None
def last_notified(self, data, reftag=None):
return None
peeringdb_server/models.py
+22
View File
@@ -945,6 +945,20 @@ class Organization(ProtectedMixin, pdb_models.OrganizationBase, GeocodeBaseMixin
help_text="Date when the organization was flagged",
)
# restrict users in the organization to be required
# to always activate 2FA.
require_2fa = models.BooleanField(
default=False,
help_text=_("Require users in your organization to activate 2FA."),
)
last_notified = models.DateTimeField(
null=True,
blank=True,
auto_now=False,
auto_now_add=False,
help_text="Date when the organization admins were notified about 2FA",
)
class Meta(pdb_models.OrganizationBase.Meta):
indexes = [models.Index(fields=["status"], name="org_status")]
@@ -5805,6 +5819,14 @@ class User(AbstractBaseUser, PermissionsMixin):
group = Group.objects.get(id=settings.USER_GROUP_ID)
return group in self.groups.all()
@property
def has_2fa(self):
"""
Returns true if the user has set up any TOTP or webauth security keys.
"""
return self.totpdevice_set.exists() or self.webauthn_security_keys.exists()
@staticmethod
def autocomplete_search_fields():
"""
peeringdb_server/org_admin_views.py
+11
View File
@@ -511,6 +511,17 @@ def uoar_approve(request, **kwargs):
uoar.delete()
return JsonResponse({"status": "ok"})
if org.require_2fa:
if not uoar.user.has_2fa:
message = (
f"User {uoar.user.full_name} requests affiliation with Organization {uoar.org.name} "
f"but has not enabled 2FA. Org {uoar.org.name} does not allow users to affiliate "
f"unless they have enabled 2FA on their account. You will be able to approve an "
f"affiliation request from User {uoar.user.full_name}, and assign permissions to "
f"them, when they have enabled 2FA."
)
return JsonResponse({"message": message}, status=403)
uoar.approve()
# notify rest of org admins that the affiliation request has been
peeringdb_server/scripts/auto_pdb_load_data.py
+61
View File
@@ -0,0 +1,61 @@
"""
Runs the pdb_load_data command on a randomized interval
Minimum interval is set through PEERINGDB_SYNC_INTERVAL environment variable (seconds)
and will default to 15 minutes if not set.
A random offset of up to 15 minutes will be appalied.
"""
import os
import random
import subprocess
import time
import structlog
# Configure logging
logger = structlog.getLogger(__name__)
def run_pdb_load_data():
command = "python manage.py pdb_load_data --commit"
directory = "/srv/www.peeringdb.com/"
subprocess.call(command, shell=True, cwd=directory)
if __name__ == "__main__":
try:
PEERINGDB_SYNC_INTERVAL = int(
os.environ.get("PEERINGDB_SYNC_INTERVAL", 15 * 60)
)
except ValueError:
PEERINGDB_SYNC_INTERVAL = 1
default_sleep = PEERINGDB_SYNC_INTERVAL
# apply a maximum offset of 15 minutes
max_sleep = default_sleep + (15 * 60)
logger.info("Starting pdb_load_data daemon...")
first_run = True
while bool(PEERINGDB_SYNC_INTERVAL):
# sleep for a random amount of time between the default interval
# and the maximum interval
if first_run:
sleep_time = 0
else:
sleep_time = random.randint(default_sleep, max_sleep)
logger.info(f"Sleeping for {sleep_time} seconds...")
time.sleep(sleep_time)
run_pdb_load_data()
logger.info("Data loaded successfully.")
first_run = False
peeringdb_server/search.py
+33
View File
@@ -242,6 +242,37 @@ def elasticsearch_proximity_entity(name):
return None
def order_results_alphabetically(result, search_term):
"""
Order the search results alphabetically and put the exact case-insensitive matches in front.
Args:
- result: A dictionary containing categories and their search results.
- search_term: A string representing the search term.
Returns:
- result: A dictionary containing the search results in alphabetical order.
"""
search_term_lower = search_term.lower()
for category in result:
result[category] = sorted(result[category], key=lambda x: x["name"].lower())
exact_match_index = -1
for index, item in enumerate(result[category]):
if item["name"].lower() == search_term_lower:
exact_match_index = index
break
if exact_match_index != -1:
exact_match = result[category].pop(exact_match_index)
result[category].insert(0, exact_match)
return result
def search_v2(term, geo={}):
"""
Search searchable objects (ixp, network, facility ...) by term on elasticsearch engine.
@@ -333,6 +364,8 @@ def search_v2(term, geo={}):
pk_map,
)
result = order_results_alphabetically(result, term)
return result
peeringdb_server/serializers.py
+5 -16
View File
@@ -283,8 +283,8 @@ class GeocodeSerializerMixin:
suggested_address = instance.process_geo_location()
# normalize state if needed
if suggested_address.get("state") != instance.state:
instance.state = suggested_address.get("state")
if suggested_address.get("state", "") != instance.state:
instance.state = suggested_address.get("state", "")
instance.save()
# provide other normalization options as suggestion to the user
@@ -1796,20 +1796,8 @@ class CarrierFacilitySerializer(ModelSerializer):
data.get("carrier"), self.Meta.model.handleref.tag
)
# new carrier-facility relationship should be pending
# until the facility approves it.
#
# if the carrier org is the same as the facility org the connection
# is auto-approved
if (
data.get("carrier")
and data.get("facility")
and data["carrier"].org_id == data["facility"].org_id
):
data["status"] = "ok"
else:
data["status"] = "pending"
# new carrier-facility relationship should be auto-approved
data["status"] = "ok"
return super().validate_create(data)
@@ -3326,6 +3314,7 @@ class OrganizationSerializer(
"website",
"social_media",
"notes",
"require_2fa",
"net_set",
"fac_set",
"ix_set",
peeringdb_server/static/20c/twentyc.edit.js
+4
View File
@@ -681,6 +681,10 @@ twentyc.editable.target.error_handlers.http_json = function(response, me, sender
info = ["Too Many Requests", response.responseJSON.message];
} else if(response.status == 403) {
info = ["Forbidden", response.responseJSON.message];
} else {
if(response.responseJSON && response.responseJSON.non_field_errors) {
peeringdb_server/static/site.css
+21 -2
View File
@@ -1645,14 +1645,22 @@ div.captcha .fallback {
padding-left: 0px;
}
.badge-2fa {
.badge-2fa-enabled {
border-radius: 3px;
background-color: orange;
background-color: green;
padding: 1px 3px;
color: #fff;
font-weight: bold;
}
.badge-2fa-disabled {
border-radius: 3px;
background-color: transparent;
padding: 1px 3px;
color: #000;
font-weight: bold;
}
/**
* right to left direction
*/
@@ -1726,3 +1734,14 @@ div.stats_row {
position: absolute;
border: 1.2px solid rgb(75, 96, 99);
}
textarea#id_redirect_uris {
width: 100%;
}
textarea.input-block-level,
input.input-block-level {
display: block;
width: 100%;
}
peeringdb_server/templates/email/notify-org-member-disable-2fa.txt
+6
View File
@@ -0,0 +1,6 @@
{% load i18n %}
{{ member }}
--------------------------------------------{% blocktrans %}
Would you like to drop their affiliation or leave it in place? If you drop affiliations, they will not be able to re-affiliate until they have 2FA turned on.
{% endblocktrans %}
[<a href="{{ drop_member_url|safe }}">{{ _('Drop Affiliations') }}</a>] [<a href="{{ leave_member_affiliate_url|safe }}">{{ _('Leave them Affiliated') }}</a>] [<a href="{{ cancel_2fa_url|safe }}">{{ _('Cancel 2FA Requirement') }}</a>]
peeringdb_server/templates/email/notify-org-user-admin-2fa.txt
+9
View File
@@ -0,0 +1,9 @@
{% load i18n %}{% blocktrans %}
The following users affiliated with {{ org_name }} organization do not have 2FA turned on:
{% endblocktrans %}
{% for member_2fa_notification in users_without_2fa %}
{{member_2fa_notification|safe}}
{% endfor %}
{% trans "You may view and edit your organization users at" %} {{ org_url }}.
{% trans "If you have questions, please don't hesitate to contact peeringdb support at" %} {{ support_email }}.
peeringdb_server/templates/site/handle-2fa.html
+22
View File
@@ -0,0 +1,22 @@
{% extends "site/base.html" %}
{% load i18n %}
{% block content %}
<div class="row">
<div class="col-12 col-md-4"></div>
<div class="panel center col-12 col-md-4">
<script>
function confirmAction() {
window.location.href = "{{ confirm_url|safe }}";
}
</script>
<h2>{% trans "Confirmation" %}</h2>
<p>{{ confirm_message }}</p>
<button class="btn btn-sm btn-success" onclick="confirmAction()">Confirm</button>
<button class="btn btn-sm btn-danger" onclick="window.location.href = '/'">Cancel</button>
</div>
<div class="col-12 col-md-4"></div>
</div>
</div>
{% endblock %}
peeringdb_server/templates/site/notification_banner.html
+10 -7
View File
@@ -1,11 +1,14 @@
{% load i18n %}
<!--
{% if not request.user.has_2fa and recent.organizations_require_2fa %}
<div class="marg-bot-15">
<div class="editable popin info panel">
<p class="center">
</p>
<div class="editable popin error">
<p class="center">
{% blocktrans trimmed %}
Some of your organizations request that you need to enable 2FA. Your access to them has been restricted until you do so.
{% endblocktrans %}
</p>
<p>{% trans "Activate 2FA for your account" %} <a href="/account/two_factor/">{% trans "here" %}</a>.</p>
</div>
</div>
</div>
-->
{% endif %}
peeringdb_server/templates/site/view_organization_tools.html
+5 -1
View File
@@ -862,7 +862,7 @@
<div class="row item editable" data-edit-id="{{ user.id }}" data-edit-label="{{ user.full_name }}">
<div class="col-4 col-sm-5 col-md-4">
<div>{{ user.full_name }}</div>
<div>{{ user.username }} {% if user.totpdevice_set.exists %}<span class="badge-2fa">2FA</span>{% endif %}{% if user.webauthn_security_keys.exists %}<span class="badge-2fa" style="margin-left:2px;">U2F</span>{% endif %}</div>
<div>{{ user.username }} {% if user.has_2fa %}<span class="badge-2fa-enabled">2FA</span>{% else %}<span class="badge-2fa-disabled">{% trans "No 2FA" %}</span>{% endif %}{% if user.webauthn_security_keys.exists %}<span class="badge-2fa" style="margin-left:2px;">U2F</span>{% endif %}</div>
</div>
<div class="col-4 col-sm-5 col-md-4">
{% for valid_email in user_emails.1 %}
@@ -907,6 +907,10 @@
<div data-edit-name="org_id">{{ instance.id }}</div>
</div>
<div class="mb-3 checkbox-inline">
<div data-edit-type="bool" data-edit-name="require_2fa" data-edit-value="{{ instance.require_2fa }}"></div>
{% trans "Require users in your organization to enable 2FA." %}
</div>
<div class="mb-3 checkbox-inline">
<div data-edit-type="bool" data-edit-name="restrict_user_emails" data-edit-value="{{ instance.restrict_user_emails }}"></div>
{% trans "Restrict user email addresses to these domains" %}
peeringdb_server/urls.py
+2
View File
@@ -47,6 +47,7 @@ from peeringdb_server.oauth_views import AuthorizationView
from peeringdb_server.views import (
OrganizationLogoUpload,
cancel_affiliation_request,
handle_2fa,
network_dismiss_ixf_proposal,
network_reset_ixf_proposals,
profile_add_email,
@@ -245,6 +246,7 @@ urlpatterns = [
peeringdb_server.org_admin_views.update_user_options,
name="org-admin-user-options",
),
re_path(r"^org_admin/handle-2fa$", handle_2fa, name="handle-2fa"),
re_path(
r"^org_admin/manage_user/delete$",
peeringdb_server.org_admin_views.manage_user_delete,
peeringdb_server/views.py
+60
View File
@@ -1302,7 +1302,15 @@ def view_index(request, errors=None):
template = loader.get_template("site/index.html")
if request.user.is_authenticated:
organizations_require_2fa = any(
org.require_2fa for org in request.user.organizations
)
else:
organizations_require_2fa = False
recent = {
"organizations_require_2fa": organizations_require_2fa,
"net": Network.handleref.filter(status="ok").order_by("-updated")[:5],
"fac": Facility.handleref.filter(status="ok").order_by("-updated")[:5],
"ix": InternetExchange.handleref.filter(status="ok").order_by("-updated")[:5],
@@ -3606,3 +3614,55 @@ def view_set_user_org(request):
response = JsonResponse({"status": "ok"})
return response
@ensure_csrf_cookie
@login_required
def handle_2fa(request):
if request.user.is_authenticated:
user_admin = request.user
action = request.GET.get("action")
commit = request.GET.get("commit")
try:
org = Organization.objects.get(id=request.GET.get("org"))
except Organization.DoesNotExist:
return view_index(request, errors=[_("Invalid organization")])
try:
member = User.objects.get(id=request.GET.get("member"))
except User.DoesNotExist:
return view_index(request, errors=[_("Invalid member")])
if user_admin.is_org_admin(org) and member.is_org_member(org):
confirm_url = f"/org_admin/handle-2fa?&org={org.id}&member={member.id}&action={action}&commit=1"
if action == "drop":
if commit:
org.usergroup.user_set.remove(member)
else:
confirm_message = f"This will completely remove {member} from {org}"
elif action == "leave":
if commit:
pass
else:
confirm_message = f"This will allow {member} to keep all privileges within {org}. This conflicts with your 2FA Policy"
elif action == "disable":
if commit:
org.require_2fa = False
org.save()
else:
confirm_message = f"This will turn off the 2FA requirement for {org}, users will not need to use 2FA to login"
else:
return view_index(request, errors=[_("Action not provided")])
if commit:
return redirect("/")
return render(
request,
"site/handle-2fa.html",
{"confirm_message": confirm_message, "confirm_url": confirm_url},
)
else:
return view_index(
request, errors=[_(f"Only admin of the {org} can perform the action")]
)
else:
return view_index(
request, errors=[_("Login as the organization admin to perform the action")]
)
poetry.lock
Generated
+637 -367
View File
File diff suppressed because it is too large Load Diff
pyproject.toml
+3 -12
View File
@@ -17,7 +17,8 @@ django-handleref = ">=2"
django-peeringdb = "==3.1.0"
djangorestframework = ">=3.14.0"
mysqlclient = ">=2.1.1"
peeringdb = ">=1.1.0, <2"
# locked to support_202306 branch until that is tagged and released to pypi
peeringdb = { git = "https://github.com/peeringdb/peeringdb-py.git", branch = "support_202306" }
uwsgi = ">=2.0.14"
# ancilary packages
bleach = ">=2.1.3"
@@ -34,17 +35,7 @@ django-extensions = ">=1.3.3"
django-grappelli = ">=2.10.1"
django-import-export = ">2.8.0,<3"
django-hashers-passlib = { git = "https://github.com/mathiasertl/django-hashers-passlib.git" }
# FIXME: django-oauth-toolkit needs to be locked to 1.6.1 because it introduced
# api breaking changes that were then reverted in 1.6.2 - in order to upgrade
# we need to wait for them to re-introduce the changes or adapt our code to
# the reversion
#
# django-oauth-toolkit 2.0.0 is ready but also introduces additional BREAKING changes that may
# need to be communicated to users, such as unreversible hashing of their current client
# secrets.
#
# See https://django-oauth-toolkit.readthedocs.io/en/latest/changelog.html#id10
django-oauth-toolkit = "==1.6.1"
django-oauth-toolkit = ">=2"
django-phonenumber-field = ">=0.6"
django-ratelimit = ">=4.0.0"
django-rest-swagger = ">=2.1.2"
tests/test_cmd_pdb_notify_2fa.py
+100
View File
@@ -0,0 +1,100 @@
from datetime import datetime, timedelta, timezone
import pytest
from django.core.management import call_command
from django_otp.plugins.otp_totp.models import TOTPDevice
from peeringdb_server.models import Organization, User
@pytest.fixture
def setup_data():
users = [
"org_admin",
"user_a",
"user_b",
"user_c",
"user_d",
"user_e",
"user_f",
]
# create test users
for name in users:
setattr(
setup_data,
name,
User.objects.create_user(name, "%s@localhost" % name, name),
)
getattr(setup_data, name).set_password(name)
# create test org
setup_data.org_required_2fa_a = Organization.objects.create(
name="Test org 2FA A", status="ok", require_2fa=True
)
setup_data.org_required_2fa_b = Organization.objects.create(
name="Test org 2FA B", status="ok", require_2fa=True
)
setup_data.org_disabled_2fa = Organization.objects.create(
name="Test org no 2FA", status="ok"
)
# add org_admin user to org as admin
setup_data.org_required_2fa_a.admin_usergroup.user_set.add(setup_data.org_admin)
# distribute group members
for user_index, user in enumerate(users):
if user_index in [1, 2]:
setup_data.org_required_2fa_a.usergroup.user_set.add(
getattr(setup_data, user)
)
elif user_index in [3, 4]:
setup_data.org_required_2fa_b.usergroup.user_set.add(
getattr(setup_data, user)
)
TOTPDevice.objects.create(user=getattr(setup_data, user), name="default")
elif user_index in [5, 6]:
setup_data.org_disabled_2fa.usergroup.user_set.add(
getattr(setup_data, user)
)
yield setup_data
@pytest.mark.django_db
def test_send_email(setup_data):
# Access the variables defined in setUpTestData
org_required_2fa_a = setup_data.org_required_2fa_a
org_required_2fa_b = setup_data.org_required_2fa_b
org_disabled_2fa = setup_data.org_disabled_2fa
MONTHS_AGO = datetime.now(tz=timezone.utc) - timedelta(days=30)
assert (
org_required_2fa_a.last_notified is None
and org_required_2fa_b.last_notified is None
and org_disabled_2fa.last_notified is None
)
# run pdb_notify_2fa command with
call_command("pdb_notify_2fa", "--commit")
org_required_2fa_a.refresh_from_db()
org_required_2fa_b.refresh_from_db()
org_disabled_2fa.refresh_from_db()
# check if last_notified is updated only for org that require 2FA has member with 2FA disabled
assert org_required_2fa_a.last_notified is not None
assert org_required_2fa_b.last_notified is None
assert org_disabled_2fa.last_notified is None
last_notified = org_required_2fa_a.last_notified
# check if last_notified is not updated if current last_notified is less than a month
call_command("pdb_notify_2fa", "--commit")
org_required_2fa_a.refresh_from_db()
assert org_required_2fa_a.last_notified == last_notified
# set the last_notified as a month has been passed
org_required_2fa_a.last_notified = MONTHS_AGO
org_required_2fa_a.save()
# check if last_notified is updated if current last_notified is more than a month
call_command("pdb_notify_2fa", "--commit")
org_required_2fa_a.refresh_from_db()
assert org_required_2fa_a.last_notified != last_notified
tests/test_orgadmin.py
+92 -2
View File
@@ -5,6 +5,7 @@ from django.conf import settings
from django.contrib.auth.models import Group
from django.test import Client, RequestFactory, TestCase
from django.urls import reverse
from django_otp.plugins.otp_totp.models import TOTPDevice
from grainy.const import *
import peeringdb_server.models as models
@@ -634,7 +635,12 @@ class OrgAdminTests(TestCase):
# test that org id was properly derived from network asn
self.assertEqual(uoar.org.id, self.org.id)
# test approval
# set org.require_2fa to True
org = self.org
org.require_2fa = True
org.save()
# test approval for user without 2FA request affiliation in organization that require 2FA
with override_group_id():
request = self.factory.post(
"/org-admin/uoar/approve?org_id=%d" % self.org.id, data={"id": uoar.id}
@@ -642,8 +648,25 @@ class OrgAdminTests(TestCase):
mock_csrf_session(request)
request.user = self.org_admin
resp = json.loads(org_admin.uoar_approve(request).content)
resp = org_admin.uoar_approve(request)
self.assertEqual(resp.status_code, 403)
self.assertEqual(
json.loads(resp.content),
{
"message": "User requests affiliation with Organization Test org but has "
"not enabled 2FA. Org Test org does not allow users to affiliate "
"unless they have enabled 2FA on their account. You will be able "
"to approve an affiliation request from User , and assign "
"permissions to them, when they have enabled 2FA."
},
)
# create user TOTP devices
totpdevice = TOTPDevice.objects.create(user=self.user_c, name="default")
totpdevice.save()
# test approval
resp = json.loads(org_admin.uoar_approve(request).content)
self.assertEqual(
{
"status": "ok",
@@ -698,6 +721,73 @@ class OrgAdminTests(TestCase):
uoar_b.delete()
def test_handle_2fa(self):
"""
Test handling a user turning off 2FA while they are in an organization that requires it
views.handle_2fa
"""
org = self.org
org_other = self.org_other
member = self.user_c
org.usergroup.user_set.add(self.user_c)
actions = ["leave", "disable", "drop"]
settings.CSRF_USE_SESSIONS = False
# check if request.user is not admin of the organization and member is not the member of the organization
request = self.factory.get(
f"/org_admin/handle-2fa?org={org_other.id}&member={member.id}&action={actions[0]}&commit=1"
)
mock_csrf_session(request)
request.user = self.org_admin
resp = views.handle_2fa(request)
self.assertEqual(resp.status_code, 200)
self.assertIn(
f"Only admin of the {org_other} can perform the action".encode(),
resp.content,
)
# confirming dialog before perform the action
for action in actions:
request = self.factory.get(
f"/org_admin/handle-2fa?org={org.id}&member={member.id}&action={action}"
)
mock_csrf_session(request)
request.user = self.org_admin
resp = views.handle_2fa(request)
self.assertEqual(resp.status_code, 200)
if action == "leave":
self.assertIn(
f"This will allow {member} to keep all privileges within {org}. This conflicts with your 2FA Policy".encode(),
resp.content,
)
if action == "disable":
self.assertIn(
f"This will turn off the 2FA requirement for {org}, users will not need to use 2FA to login".encode(),
resp.content,
)
if action == "drop":
self.assertIn(
f"This will completely remove {member} from {org}".encode(),
resp.content,
)
# after agree in the confirming dialog
for action in actions:
request = self.factory.get(
f"/org_admin/handle-2fa?org={org.id}&member={member.id}&action={action}&commit=1"
)
mock_csrf_session(request)
request.user = self.org_admin
resp = views.handle_2fa(request)
self.assertEqual(resp.status_code, 302)
if action == "leave":
self.assertIn(member, org.usergroup.user_set.all())
if action == "disable":
self.assertFalse(org.require_2fa)
if action == "drop":
self.assertNotIn(member, org.usergroup.user_set.all())
settings.CSRF_USE_SESSIONS = True
def test_uoar_deny(self):
"""
Test denying of a user-org-affiliation-request
tests/util.py
+2 -2
View File
@@ -5,7 +5,7 @@ from importlib import import_module
from django.conf import settings
from django.contrib.auth.models import AnonymousUser, Group
from django.contrib.sessions.models import Session
from django.middleware.csrf import CSRF_SESSION_KEY
from django.middleware.csrf import CSRF_SESSION_KEY, _get_new_csrf_string
from django.test import TestCase
from django_grainy.models import GroupPermission, UserPermission
@@ -135,5 +135,5 @@ def setup_test_data(filename):
def mock_csrf_session(request):
engine = import_module(settings.SESSION_ENGINE)
request.session = engine.SessionStore("deadbeef")
request.session[CSRF_SESSION_KEY] = "csrf-session-key"
request.session[CSRF_SESSION_KEY] = _get_new_csrf_string()
request._dont_enforce_csrf_checks = True