feat: create_oidc management command to create SSO integration for
ci/woodpecker/push/woodpecker Pipeline was successful
Details
ci/woodpecker/push/woodpecker Pipeline was successful
Details
Hosteawip-payments
parent
cc1ecaf6a0
commit
8507129063
|
@ -0,0 +1,110 @@
|
||||||
|
# Copyright © 2022 Aravinth Manivannan <realaravinth@batsense.net>
|
||||||
|
#
|
||||||
|
# This program is free software: you can redistribute it and/or modify
|
||||||
|
# it under the terms of the GNU Affero General Public License as
|
||||||
|
# published by the Free Software Foundation, either version 3 of the
|
||||||
|
# License, or (at your option) any later version.
|
||||||
|
#
|
||||||
|
# This program is distributed in the hope that it will be useful,
|
||||||
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||||
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||||
|
# GNU Affero General Public License for more details.
|
||||||
|
#
|
||||||
|
# You should have received a copy of the GNU Affero General Public License
|
||||||
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
from django.core.management.base import BaseCommand
|
||||||
|
from django.core.exceptions import ValidationError
|
||||||
|
from django.conf import settings
|
||||||
|
from django.contrib.auth import get_user_model
|
||||||
|
|
||||||
|
from oauth2_provider.models import get_application_model
|
||||||
|
from oauth2_provider.generators import generate_client_id, generate_client_secret
|
||||||
|
|
||||||
|
from accounts.utils import gen_secret
|
||||||
|
|
||||||
|
Application = get_application_model()
|
||||||
|
|
||||||
|
|
||||||
|
class Command(BaseCommand):
|
||||||
|
help = "Get user ID from username"
|
||||||
|
app_name_key = "app_name"
|
||||||
|
username_key = "username"
|
||||||
|
redirect_uri_key = "redirect_uri"
|
||||||
|
|
||||||
|
def add_arguments(self, parser):
|
||||||
|
parser.add_argument(self.app_name_key, type=str, help="The application name")
|
||||||
|
|
||||||
|
parser.add_argument(
|
||||||
|
self.username_key,
|
||||||
|
type=str,
|
||||||
|
help="The username of user who will own this app",
|
||||||
|
)
|
||||||
|
|
||||||
|
parser.add_argument(
|
||||||
|
self.redirect_uri_key,
|
||||||
|
type=str,
|
||||||
|
help="The username of user who will own this app",
|
||||||
|
)
|
||||||
|
|
||||||
|
def handle(self, *args, **options):
|
||||||
|
if self.username_key not in options:
|
||||||
|
self.stdout.write(self.style.ERROR("Please provide username"))
|
||||||
|
return
|
||||||
|
if self.app_name_key not in options:
|
||||||
|
self.stdout.write(self.style.ERROR("Please provide application name"))
|
||||||
|
return
|
||||||
|
|
||||||
|
if self.redirect_uri_key not in options:
|
||||||
|
self.stdout.write(self.style.ERROR("Please provide redirect uri"))
|
||||||
|
return
|
||||||
|
|
||||||
|
username = options[self.username_key]
|
||||||
|
application_name = options[self.app_name_key]
|
||||||
|
redirect_uri = options[self.redirect_uri_key]
|
||||||
|
|
||||||
|
User = get_user_model()
|
||||||
|
if not User.objects.filter(username=username).exists():
|
||||||
|
self.stderr.write(self.style.ERROR(f"user {username} not found"))
|
||||||
|
return
|
||||||
|
|
||||||
|
user = User.objects.get(username=username)
|
||||||
|
# python manage.py createapplication --name demo-oidc-app --user 1 --client-id 22500acb0bcfcba137d6b8ae96d3f2 --client-secret 296055337620b0e443ad24a32cb675 --algorithm HS256 --skip-authorization --redirect-uri http://example.org/uri1 confidential code -v
|
||||||
|
|
||||||
|
client_id = generate_client_id()
|
||||||
|
client_secret = generate_client_secret()
|
||||||
|
config = {
|
||||||
|
"name": application_name,
|
||||||
|
"user_id": user.id,
|
||||||
|
"client_id": client_id,
|
||||||
|
"client_secret": client_secret,
|
||||||
|
"algorithm": "HS256",
|
||||||
|
"skip_authorization": True,
|
||||||
|
"redirect_uris": redirect_uri,
|
||||||
|
"authorization_grant_type": "authorization-code",
|
||||||
|
"client_type": "confidential",
|
||||||
|
}
|
||||||
|
|
||||||
|
app = Application(**config)
|
||||||
|
|
||||||
|
try:
|
||||||
|
app.full_clean()
|
||||||
|
except ValidationError as exc:
|
||||||
|
errors = "\n ".join(
|
||||||
|
[
|
||||||
|
"- " + err_key + ": " + str(err_value)
|
||||||
|
for err_key, err_value in exc.message_dict.items()
|
||||||
|
]
|
||||||
|
)
|
||||||
|
self.stdout.write(
|
||||||
|
self.style.ERROR("Please correct the following errors:\n %s" % errors)
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
app.save()
|
||||||
|
self.stdout.write(
|
||||||
|
self.style.SUCCESS(
|
||||||
|
f"New application {application_name} created successfully."
|
||||||
|
)
|
||||||
|
)
|
||||||
|
self.stdout.write(f"client_id: {client_id}")
|
||||||
|
self.stdout.write(f"client_secret: {client_secret}")
|
|
@ -1,43 +0,0 @@
|
||||||
# Copyright © 2022 Aravinth Manivannan <realaravinth@batsense.net>
|
|
||||||
#
|
|
||||||
# This program is free software: you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU Affero General Public License as
|
|
||||||
# published by the Free Software Foundation, either version 3 of the
|
|
||||||
# License, or (at your option) any later version.
|
|
||||||
#
|
|
||||||
# This program is distributed in the hope that it will be useful,
|
|
||||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
||||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
||||||
# GNU Affero General Public License for more details.
|
|
||||||
#
|
|
||||||
# You should have received a copy of the GNU Affero General Public License
|
|
||||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
||||||
from datetime import datetime, timedelta
|
|
||||||
from django.core.management.base import BaseCommand
|
|
||||||
from django.conf import settings
|
|
||||||
from django.contrib.auth import get_user_model
|
|
||||||
|
|
||||||
|
|
||||||
class Command(BaseCommand):
|
|
||||||
help = "Get user ID from username"
|
|
||||||
username_key = "username"
|
|
||||||
|
|
||||||
def add_arguments(self, parser):
|
|
||||||
parser.add_argument(
|
|
||||||
self.username_key,
|
|
||||||
type=str,
|
|
||||||
help="The username for which database assigned ID needs to be queried",
|
|
||||||
)
|
|
||||||
|
|
||||||
def handle(self, *args, **options):
|
|
||||||
if self.username_key not in options:
|
|
||||||
self.stdout.write(self.style.ERROR("Please provide username"))
|
|
||||||
return
|
|
||||||
username = options[self.username_key]
|
|
||||||
User = get_user_model()
|
|
||||||
if User.objects.filter(username=username).exists():
|
|
||||||
user = get_user_model().objects.get(username=username)
|
|
||||||
self.stdout.write(self.style.SUCCESS(user.id))
|
|
||||||
return
|
|
||||||
else:
|
|
||||||
self.stderr.write(self.style.ERROR(f"user {username} not found"))
|
|
|
@ -25,6 +25,9 @@ from django.utils.http import urlencode
|
||||||
from django.contrib.auth import authenticate
|
from django.contrib.auth import authenticate
|
||||||
from django.conf import settings
|
from django.conf import settings
|
||||||
|
|
||||||
|
|
||||||
|
from oauth2_provider.models import get_application_model
|
||||||
|
|
||||||
from .models import AccountConfirmChallenge
|
from .models import AccountConfirmChallenge
|
||||||
from .management.commands.rm_unverified_users import (
|
from .management.commands.rm_unverified_users import (
|
||||||
Command as CleanUnverifiedUsersCommand,
|
Command as CleanUnverifiedUsersCommand,
|
||||||
|
@ -380,19 +383,49 @@ class ConfirmAccessDecorator(TestCase):
|
||||||
self.assertEqual(fn(req), True)
|
self.assertEqual(fn(req), True)
|
||||||
|
|
||||||
|
|
||||||
class GetUserIDTest(TestCase):
|
class CreateOidCApplicaiton(TestCase):
|
||||||
|
"""
|
||||||
|
Test command: manage.py create_oidc
|
||||||
|
"""
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.username = "getuserid"
|
self.username = "oidcadmin"
|
||||||
register_util(t=self, username=self.username)
|
register_util(t=self, username=self.username)
|
||||||
|
|
||||||
def test_command_output(self):
|
def test_cmd(self):
|
||||||
|
|
||||||
|
Application = get_application_model()
|
||||||
|
|
||||||
stdout = StringIO()
|
stdout = StringIO()
|
||||||
stderr = StringIO()
|
stderr = StringIO()
|
||||||
# username exists
|
|
||||||
call_command(f"get_user_id {self.username}", stdout=stdout, stderr=stderr)
|
|
||||||
self.assertIn(self.user.id, out.getvalue())
|
|
||||||
|
|
||||||
# username doesn't exist
|
redirect_uri = "http://example.org"
|
||||||
nouser = "fooabr"
|
app_name = "test_cmd_oidc"
|
||||||
call_command(f"get_user_id {nouser}", stdout=out, stderr=stderr)
|
|
||||||
self.assertIn(f"username {nouser} doesn't exist", stderr.getvalue())
|
# username exists
|
||||||
|
call_command(
|
||||||
|
"create_oidc",
|
||||||
|
app_name,
|
||||||
|
self.username,
|
||||||
|
redirect_uri,
|
||||||
|
stdout=stdout,
|
||||||
|
stderr=stderr,
|
||||||
|
)
|
||||||
|
out = stdout.getvalue()
|
||||||
|
|
||||||
|
self.assertIn(f"New application {app_name} created successfully", out)
|
||||||
|
|
||||||
|
client_id = out.split("\n")[1].split(" ")[1]
|
||||||
|
|
||||||
|
self.assertEqual(
|
||||||
|
get_application_model().objects.filter(client_id=client_id).exists(), True
|
||||||
|
)
|
||||||
|
app = get_application_model().objects.get(name=app_name)
|
||||||
|
self.assertEqual(app.client_id, client_id)
|
||||||
|
self.assertEqual(app.name, app_name)
|
||||||
|
self.assertEqual(app.user_id, self.user.id)
|
||||||
|
self.assertEqual(app.redirect_uris, redirect_uri)
|
||||||
|
self.assertEqual(app.skip_authorization, True)
|
||||||
|
self.assertEqual(app.client_type, "confidential")
|
||||||
|
self.assertEqual(app.authorization_grant_type, "authorization-code")
|
||||||
|
self.assertEqual(app.algorithm, "HS256")
|
||||||
|
|
Loading…
Reference in New Issue