python/samba/netcmd/user/auth/silo.py

   1 # Unix SMB/CIFS implementation.
   2 #
   3 # manage assigned authentication silos on a user
   4 #
   5 # Copyright (C) Catalyst.Net Ltd. 2023
   6 #
   7 # Written by Rob van der Linde <rob@catalyst.net.nz>
   8 #
   9 # This program is free software; you can redistribute it and/or modify
  10 # it under the terms of the GNU General Public License as published by
  11 # the Free Software Foundation; either version 3 of the License, or
  12 # (at your option) any later version.
  13 #
  14 # This program is distributed in the hope that it will be useful,
  15 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  16 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17 # GNU General Public License for more details.
  18 #
  19 # You should have received a copy of the GNU General Public License
  20 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  21 #
  22
  23 import samba.getopt as options
  24 from samba.domain.models import AuthenticationSilo, User
  25 from samba.domain.models.exceptions import ModelError
  26 from samba.netcmd import Command, CommandError, Option, SuperCommand
  27
  28
  29 class cmd_user_auth_silo_assign(Command):
  30     """Set the assigned authentication silo on a user."""
  31
  32     synopsis = "%prog <username> [options]"
  33
  34     takes_args = ["username"]
  35
  36     takes_optiongroups = {
  37         "sambaopts": options.SambaOptions,
  38         "credopts": options.CredentialsOptions,
  39         "hostopts": options.HostOptions,
  40     }
  41
  42     takes_options = [
  43         Option("--silo", help="Authentication silo name.",
  44                action="store", dest="silo_name", type=str, required=True),
  45     ]
  46
  47     def run(self, username, hostopts=None, sambaopts=None, credopts=None,
  48             silo_name=None):
  49
  50         ldb = self.ldb_connect(hostopts, sambaopts, credopts)
  51
  52         try:
  53             user = User.find(ldb, username)
  54             silo = AuthenticationSilo.get(ldb, name=silo_name)
  55         except ModelError as e:
  56             raise CommandError(e)
  57
  58         # User and silo exist.
  59         if user is None:
  60             raise CommandError(f"User {username} not found.")
  61         if silo is None:
  62             raise CommandError(f"Authentication silo {silo_name} not found.")
  63
  64         # Set assigned silo.
  65         user.assigned_silo = silo.dn
  66
  67         try:
  68             user.save(ldb)
  69         except ModelError as e:
  70             raise CommandError(f"Set assigned authentication silo failed: {e}")
  71
  72         # Display silo member status.
  73         if user.dn in silo.members:
  74             status = "granted"
  75         else:
  76             status = "revoked"
  77
  78         print(f"User {username} assigned to authentication silo {silo} ({status})",
  79               file=self.outf)
  80
  81
  82 class cmd_user_auth_silo_remove(Command):
  83     """Remove the assigned authentication silo on a user."""
  84
  85     synopsis = "%prog <username> [options]"
  86
  87     takes_args = ["username"]
  88
  89     takes_optiongroups = {
  90         "sambaopts": options.SambaOptions,
  91         "credopts": options.CredentialsOptions,
  92         "hostopts": options.HostOptions,
  93     }
  94
  95     def run(self, username, hostopts=None, sambaopts=None, credopts=None):
  96
  97         ldb = self.ldb_connect(hostopts, sambaopts, credopts)
  98
  99         try:
 100             user = User.find(ldb, username)
 101         except ModelError as e:
 102             raise CommandError(e)
 103
 104         # User exists
 105         if user is None:
 106             raise CommandError(f"User {username} not found.")
 107
 108         # Get previous silo for display.
 109         if user.assigned_silo:
 110             try:
 111                 silo = AuthenticationSilo.get(ldb, dn=user.assigned_silo)
 112             except ModelError as e:
 113                 raise CommandError(e)
 114         else:
 115             silo = None
 116
 117         # Unset assigned authentication silo
 118         user.assigned_silo = None
 119
 120         try:
 121             user.save(ldb)
 122         except ModelError as e:
 123             raise CommandError(f"Remove assigned authentication silo failed: {e}")
 124
 125         # Display silo member status.
 126         if silo and user.dn in silo.members:
 127             status = "granted"
 128         else:
 129             status = "revoked"
 130
 131         print(f"User {username} removed from authentication silo {silo} ({status})",
 132               file=self.outf)
 133
 134
 135 class cmd_user_auth_silo_view(Command):
 136     """View the current assigned authentication silo on a user."""
 137
 138     synopsis = "%prog <username> [options]"
 139
 140     takes_args = ["username"]
 141
 142     takes_optiongroups = {
 143         "sambaopts": options.SambaOptions,
 144         "credopts": options.CredentialsOptions,
 145         "hostopts": options.HostOptions,
 146     }
 147
 148     def run(self, username, hostopts=None, sambaopts=None, credopts=None):
 149
 150         ldb = self.ldb_connect(hostopts, sambaopts, credopts)
 151
 152         try:
 153             user = User.find(ldb, username)
 154
 155             # Check user exists before fetching silo.
 156             if user is None:
 157                 raise CommandError(f"User {username} not found.")
 158
 159             # Only fetch silo is one is assigned.
 160             if user.assigned_silo:
 161                 silo = AuthenticationSilo.get(ldb, dn=user.assigned_silo)
 162             else:
 163                 silo = None
 164
 165         except ModelError as e:
 166             raise CommandError(e)
 167
 168         # Display silo member status.
 169         if silo and user.dn in silo.members:
 170             status = "granted"
 171         else:
 172             status = "revoked"
 173
 174         if silo:
 175             print(f"User {username} assigned to authentication silo {silo} ({status})",
 176                   file=self.outf)
 177         else:
 178             print(f"User {username} has no assigned authentication silo.",
 179                   file=self.outf)
 180
 181
 182 class cmd_user_auth_silo(SuperCommand):
 183     """Manage authentication silos on a user."""
 184
 185     subcommands = {
 186         "assign": cmd_user_auth_silo_assign(),
 187         "remove": cmd_user_auth_silo_remove(),
 188         "view": cmd_user_auth_silo_view(),
 189     }