authentik.core.management.commands.shell

authentik shell command

 1"""authentik shell command"""
 2
 3import platform
 4from pprint import pprint
 5
 6from django.core.management.commands.shell import Command as BaseCommand
 7from django.db.models import Model
 8from django.db.models.signals import post_save, pre_delete
 9
10from authentik import authentik_full_version
11from authentik.core.models import User
12from authentik.events.middleware import should_log_model
13from authentik.events.models import Event, EventAction
14from authentik.events.utils import model_to_dict
15
16
17def get_banner_text(shell_type="shell") -> str:
18    return f"""### authentik {shell_type} ({authentik_full_version()})
19### Node {platform.node()} | Arch {platform.machine()} | Python {platform.python_version()} """
20
21
22class Command(BaseCommand):
23    """Start the Django shell with all authentik models already imported"""
24
25    def get_namespace(self, **options):
26        return {
27            **super().get_namespace(**options),
28            "pprint": pprint,
29        }
30
31    @staticmethod
32    def post_save_handler(sender, instance: Model, created: bool, **_):
33        """Signal handler for all object's post_save"""
34        if not should_log_model(instance):
35            return
36
37        action = EventAction.MODEL_CREATED if created else EventAction.MODEL_UPDATED
38        Event.new(action, model=model_to_dict(instance)).set_user(
39            User(
40                username="authentik-shell",
41                pk=0,
42                email="",
43            )
44        ).save()
45
46    @staticmethod
47    def pre_delete_handler(sender, instance: Model, **_):
48        """Signal handler for all object's pre_delete"""
49        if not should_log_model(instance):  # pragma: no cover
50            return
51
52        Event.new(EventAction.MODEL_DELETED, model=model_to_dict(instance)).set_user(
53            User(
54                username="authentik-shell",
55                pk=0,
56                email="",
57            )
58        ).save()
59
60    def handle(self, **options):
61        post_save.connect(Command.post_save_handler)
62        pre_delete.connect(Command.pre_delete_handler)
63
64        print(get_banner_text())
65
66        super().handle(**options)
def get_banner_text(shell_type='shell') -> str: 
18def get_banner_text(shell_type="shell") -> str:
19    return f"""### authentik {shell_type} ({authentik_full_version()})
20### Node {platform.node()} | Arch {platform.machine()} | Python {platform.python_version()} """
class Command(django.core.management.commands.shell.Command): 
23class Command(BaseCommand):
24    """Start the Django shell with all authentik models already imported"""
25
26    def get_namespace(self, **options):
27        return {
28            **super().get_namespace(**options),
29            "pprint": pprint,
30        }
31
32    @staticmethod
33    def post_save_handler(sender, instance: Model, created: bool, **_):
34        """Signal handler for all object's post_save"""
35        if not should_log_model(instance):
36            return
37
38        action = EventAction.MODEL_CREATED if created else EventAction.MODEL_UPDATED
39        Event.new(action, model=model_to_dict(instance)).set_user(
40            User(
41                username="authentik-shell",
42                pk=0,
43                email="",
44            )
45        ).save()
46
47    @staticmethod
48    def pre_delete_handler(sender, instance: Model, **_):
49        """Signal handler for all object's pre_delete"""
50        if not should_log_model(instance):  # pragma: no cover
51            return
52
53        Event.new(EventAction.MODEL_DELETED, model=model_to_dict(instance)).set_user(
54            User(
55                username="authentik-shell",
56                pk=0,
57                email="",
58            )
59        ).save()
60
61    def handle(self, **options):
62        post_save.connect(Command.post_save_handler)
63        pre_delete.connect(Command.pre_delete_handler)
64
65        print(get_banner_text())
66
67        super().handle(**options)

Start the Django shell with all authentik models already imported

def get_namespace(self, **options): 
26    def get_namespace(self, **options):
27        return {
28            **super().get_namespace(**options),
29            "pprint": pprint,
30        }
@staticmethod
def post_save_handler(sender, instance: django.db.models.base.Model, created: bool, **_): 
32    @staticmethod
33    def post_save_handler(sender, instance: Model, created: bool, **_):
34        """Signal handler for all object's post_save"""
35        if not should_log_model(instance):
36            return
37
38        action = EventAction.MODEL_CREATED if created else EventAction.MODEL_UPDATED
39        Event.new(action, model=model_to_dict(instance)).set_user(
40            User(
41                username="authentik-shell",
42                pk=0,
43                email="",
44            )
45        ).save()

Signal handler for all object's post_save

@staticmethod
def pre_delete_handler(sender, instance: django.db.models.base.Model, **_): 
47    @staticmethod
48    def pre_delete_handler(sender, instance: Model, **_):
49        """Signal handler for all object's pre_delete"""
50        if not should_log_model(instance):  # pragma: no cover
51            return
52
53        Event.new(EventAction.MODEL_DELETED, model=model_to_dict(instance)).set_user(
54            User(
55                username="authentik-shell",
56                pk=0,
57                email="",
58            )
59        ).save()

Signal handler for all object's pre_delete

def handle(self, **options): 
61    def handle(self, **options):
62        post_save.connect(Command.post_save_handler)
63        pre_delete.connect(Command.pre_delete_handler)
64
65        print(get_banner_text())
66
67        super().handle(**options)

The actual logic of the command. Subclasses must implement this method.