sin/samba-mirror
1
0
Fork 0
mirror of https://github.com/samba-team/samba.git synced 2025-03-11 16:58:40 +03:00
Code

netcmd: add auth silo and policy sub-commands to samba-tool user

Browse Source 
Signed-off-by: Rob van der Linde <rob@catalyst.net.nz>
Reviewed-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>
This commit is contained in:
Rob van der Linde 2023-10-30 12:43:57 +13:00 committed by Andrew Bartlett
parent 2f20fa9b90
commit 422680f82f
4 changed files with 396 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
python/samba/netcmd/user/__init__.py
View File

@ -21,6 +21,7 @@ from samba.netcmd import SuperCommand
from .add import cmd_user_add
from .add_unix_attrs import cmd_user_add_unix_attrs
from .auth import cmd_user_auth
from .delete import cmd_user_delete
from .disable import cmd_user_disable
from .edit import cmd_user_edit
@ -43,6 +44,7 @@ class cmd_user(SuperCommand):
"""User management."""
subcommands = {}
subcommands["auth"] = cmd_user_auth()
subcommands["add"] = cmd_user_add()
subcommands["create"] = cmd_user_add()
subcommands["delete"] = cmd_user_delete()

35
python/samba/netcmd/user/auth/__init__.py Normal file
View File

@ -0,0 +1,35 @@
# Unix SMB/CIFS implementation.
#
# manage assigned authentication policies and silos on a user
#
# Copyright (C) Catalyst.Net Ltd. 2023
#
# Written by Rob van der Linde <rob@catalyst.net.nz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from samba.netcmd import SuperCommand
from .policy import cmd_user_auth_policy
from .silo import cmd_user_auth_silo
class cmd_user_auth(SuperCommand):
"""Manage authentication policies and silos on a user."""
subcommands = {
"policy": cmd_user_auth_policy(),
"silo": cmd_user_auth_silo(),
}

170
python/samba/netcmd/user/auth/policy.py Normal file
View File

@ -0,0 +1,170 @@
# Unix SMB/CIFS implementation.
#
# manage assigned authentication policies on a user
#
# Copyright (C) Catalyst.Net Ltd. 2023
#
# Written by Rob van der Linde <rob@catalyst.net.nz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import samba.getopt as options
from samba.netcmd import Command, CommandError, Option, SuperCommand
from samba.netcmd.domain.models import AuthenticationPolicy, User
from samba.netcmd.domain.models.exceptions import ModelError
class cmd_user_auth_policy_assign(Command):
"""Set the assigned authentication policy on a user."""
synopsis = "%prog <username> [options]"
takes_args = ["username"]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
"credopts": options.CredentialsOptions,
"hostopts": options.HostOptions,
}
takes_options = [
Option("--policy", help="Authentication policy name.",
action="store", dest="policy_name", type=str, required=True),
]
def run(self, username, hostopts=None, sambaopts=None, credopts=None,
policy_name=None):
ldb = self.ldb_connect(hostopts, sambaopts, credopts)
try:
user = User.find(ldb, username)
policy = AuthenticationPolicy.get(ldb, name=policy_name)
except ModelError as e:
raise CommandError(e)
# User and policy exist.
if user is None:
raise CommandError(f"User {username} not found.")
if policy is None:
raise CommandError(f"Authentication policy {policy_name} not found.")
# Set assigned policy.
user.assigned_policy = policy.dn
try:
user.save(ldb)
except ModelError as e:
raise CommandError(f"Set assigned authentication policy failed: {e}")
print(f"User {username} assigned to authentication policy {policy}",
file=self.outf)
class cmd_user_auth_policy_remove(Command):
"""Remove the assigned authentication policy on a user."""
synopsis = "%prog <username> [options]"
takes_args = ["username"]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
"credopts": options.CredentialsOptions,
"hostopts": options.HostOptions,
}
def run(self, username, hostopts=None, sambaopts=None, credopts=None):
ldb = self.ldb_connect(hostopts, sambaopts, credopts)
try:
user = User.find(ldb, username)
except ModelError as e:
raise CommandError(e)
# User exists
if user is None:
raise CommandError(f"User {username} not found.")
# Get previous policy for display.
if user.assigned_policy:
try:
policy = AuthenticationPolicy.get(ldb, dn=user.assigned_policy)
except ModelError as e:
raise CommandError(e)
else:
policy = None
# Unset assigned authentication policy
user.assigned_policy = None
try:
user.save(ldb)
except ModelError as e:
raise CommandError(f"Remove assigned authentication policy failed: {e}")
print(f"User {username} removed from authentication policy {policy}",
file=self.outf)
class cmd_user_auth_policy_view(Command):
"""View the current assigned authentication policy on a user."""
synopsis = "%prog <username> [options]"
takes_args = ["username"]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
"credopts": options.CredentialsOptions,
"hostopts": options.HostOptions,
}
def run(self, username, hostopts=None, sambaopts=None, credopts=None):
ldb = self.ldb_connect(hostopts, sambaopts, credopts)
try:
user = User.find(ldb, username)
# Check user exists before fetching policy.
if user is None:
raise CommandError(f"User {username} not found.")
if user.assigned_policy:
policy = AuthenticationPolicy.get(ldb, dn=user.assigned_policy)
else:
policy = None
except ModelError as e:
raise CommandError(e)
if policy:
print(f"User {username} assigned to authentication policy {policy}",
file=self.outf)
else:
print(f"User {username} has no assigned authentication policy.",
file=self.outf)
class cmd_user_auth_policy(SuperCommand):
"""Manage authentication policies on a user."""
subcommands = {
"assign": cmd_user_auth_policy_assign(),
"remove": cmd_user_auth_policy_remove(),
"view": cmd_user_auth_policy_view(),
}

189
python/samba/netcmd/user/auth/silo.py Normal file
View File

@ -0,0 +1,189 @@
# Unix SMB/CIFS implementation.
#
# manage assigned authentication silos on a user
#
# Copyright (C) Catalyst.Net Ltd. 2023
#
# Written by Rob van der Linde <rob@catalyst.net.nz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import samba.getopt as options
from samba.netcmd import Command, CommandError, Option, SuperCommand
from samba.netcmd.domain.models import AuthenticationSilo, User
from samba.netcmd.domain.models.exceptions import ModelError
class cmd_user_auth_silo_assign(Command):
"""Set the assigned authentication silo on a user."""
synopsis = "%prog <username> [options]"
takes_args = ["username"]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
"credopts": options.CredentialsOptions,
"hostopts": options.HostOptions,
}
takes_options = [
Option("--silo", help="Authentication silo name.",
action="store", dest="silo_name", type=str, required=True),
]
def run(self, username, hostopts=None, sambaopts=None, credopts=None,
silo_name=None):
ldb = self.ldb_connect(hostopts, sambaopts, credopts)
try:
user = User.find(ldb, username)
silo = AuthenticationSilo.get(ldb, name=silo_name)
except ModelError as e:
raise CommandError(e)
# User and silo exist.
if user is None:
raise CommandError(f"User {username} not found.")
if silo is None:
raise CommandError(f"Authentication silo {silo_name} not found.")
# Set assigned silo.
user.assigned_silo = silo.dn
try:
user.save(ldb)
except ModelError as e:
raise CommandError(f"Set assigned authentication silo failed: {e}")
# Display silo member status.
if user.dn in silo.members:
status = "granted"
else:
status = "revoked"
print(f"User {username} assigned to authentication silo {silo} ({status})",
file=self.outf)
class cmd_user_auth_silo_remove(Command):
"""Remove the assigned authentication silo on a user."""
synopsis = "%prog <username> [options]"
takes_args = ["username"]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
"credopts": options.CredentialsOptions,
"hostopts": options.HostOptions,
}
def run(self, username, hostopts=None, sambaopts=None, credopts=None):
ldb = self.ldb_connect(hostopts, sambaopts, credopts)
try:
user = User.find(ldb, username)
except ModelError as e:
raise CommandError(e)
# User exists
if user is None:
raise CommandError(f"User {username} not found.")
# Get previous silo for display.
if user.assigned_silo:
try:
silo = AuthenticationSilo.get(ldb, dn=user.assigned_silo)
except ModelError as e:
raise CommandError(e)
else:
silo = None
# Unset assigned authentication silo
user.assigned_silo = None
try:
user.save(ldb)
except ModelError as e:
raise CommandError(f"Remove assigned authentication silo failed: {e}")
# Display silo member status.
if silo and user.dn in silo.members:
status = "granted"
else:
status = "revoked"
print(f"User {username} removed from authentication silo {silo} ({status})",
file=self.outf)
class cmd_user_auth_silo_view(Command):
"""View the current assigned authentication silo on a user."""
synopsis = "%prog <username> [options]"
takes_args = ["username"]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
"credopts": options.CredentialsOptions,
"hostopts": options.HostOptions,
}
def run(self, username, hostopts=None, sambaopts=None, credopts=None):
ldb = self.ldb_connect(hostopts, sambaopts, credopts)
try:
user = User.find(ldb, username)
# Check user exists before fetching silo.
if user is None:
raise CommandError(f"User {username} not found.")
# Only fetch silo is one is assigned.
if user.assigned_silo:
silo = AuthenticationSilo.get(ldb, dn=user.assigned_silo)
else:
silo = None
except ModelError as e:
raise CommandError(e)
# Display silo member status.
if silo and user.dn in silo.members:
status = "granted"
else:
status = "revoked"
if silo:
print(f"User {username} assigned to authentication silo {silo} ({status})",
file=self.outf)
else:
print(f"User {username} has no assigned authentication silo.",
file=self.outf)
class cmd_user_auth_silo(SuperCommand):
"""Manage authentication silos on a user."""
subcommands = {
"assign": cmd_user_auth_silo_assign(),
"remove": cmd_user_auth_silo_remove(),
"view": cmd_user_auth_silo_view(),
}