Page Menu
Home
VyOS Platform
Search
Configure Global Search
Log In
Files
F118307128
No One
Temporary
Actions
View File
Edit File
Delete File
View Transforms
Subscribe
Flag For Later
Award Token
Size
23 KB
Referenced Files
None
Subscribers
None
View Options
diff --git a/smoketest/scripts/cli/test_system_login.py b/smoketest/scripts/cli/test_system_login.py
index 8327235fb..af3a5851c 100755
--- a/smoketest/scripts/cli/test_system_login.py
+++ b/smoketest/scripts/cli/test_system_login.py
@@ -1,204 +1,239 @@
#!/usr/bin/env python3
#
# Copyright (C) 2019-2020 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# 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 re
import platform
import unittest
from base_vyostest_shim import VyOSUnitTestSHIM
from distutils.version import LooseVersion
from platform import release as kernel_version
from subprocess import Popen, PIPE
from vyos.configsession import ConfigSession
from vyos.configsession import ConfigSessionError
from vyos.util import cmd
from vyos.util import read_file
from vyos.template import inc_ip
base_path = ['system', 'login']
-users = ['vyos1', 'vyos2']
+users = ['vyos1', 'vyos-roxx123', 'VyOS-123_super.Nice']
+
+ssh_pubkey = """
+AAAAB3NzaC1yc2EAAAADAQABAAABgQD0NuhUOEtMIKnUVFIHoFatqX/c4mjerXyF
+TlXYfVt6Ls2NZZsUSwHbnhK4BKDrPvVZMW/LycjQPzWW6TGtk6UbZP1WqdviQ9hP
+jsEeKJSTKciMSvQpjBWyEQQPXSKYQC7ryQQilZDqnJgzqwzejKEe+nhhOdBvjuZc
+uukxjT69E0UmWAwLxzvfiurwiQaC7tG+PwqvtfHOPL3i6yRO2C5ORpFarx8PeGDS
+IfIXJCr3LoUbLHeuE7T2KaOKQcX0UsWJ4CoCapRLpTVYPDB32BYfgq7cW1Sal1re
+EGH2PzuXBklinTBgCHA87lHjpwDIAqdmvMj7SXIW9LxazLtP+e37sexE7xEs0cpN
+l68txdDbY2P2Kbz5mqGFfCvBYKv9V2clM5vyWNy/Xp5TsCis89nn83KJmgFS7sMx
+pHJz8umqkxy3hfw0K7BRFtjWd63sbOP8Q/SDV7LPaIfIxenA9zv2rY7y+AIqTmSr
+TTSb0X1zPGxPIRFy5GoGtO9Mm5h4OZk=
+"""
class TestSystemLogin(VyOSUnitTestSHIM.TestCase):
def tearDown(self):
# Delete individual users from configuration
for user in users:
self.cli_delete(base_path + ['user', user])
self.cli_commit()
def test_add_linux_system_user(self):
+ # We are not allowed to re-use a username already taken by the Linux
+ # base system
system_user = 'backup'
self.cli_set(base_path + ['user', system_user, 'authentication', 'plaintext-password', system_user])
# check validate() - can not add username which exists on the Debian
# base system (UID < 1000)
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_delete(base_path + ['user', system_user])
def test_system_login_user(self):
# Check if user can be created and we can SSH to localhost
self.cli_set(['service', 'ssh', 'port', '22'])
for user in users:
name = "VyOS Roxx " + user
home_dir = "/tmp/" + user
self.cli_set(base_path + ['user', user, 'authentication', 'plaintext-password', user])
self.cli_set(base_path + ['user', user, 'full-name', 'VyOS Roxx'])
self.cli_set(base_path + ['user', user, 'home-directory', home_dir])
self.cli_commit()
for user in users:
cmd = ['su','-', user]
proc = Popen(cmd, stdin=PIPE, stdout=PIPE, stderr=PIPE)
tmp = "{}\nuname -a".format(user)
proc.stdin.write(tmp.encode())
proc.stdin.flush()
(stdout, stderr) = proc.communicate()
# stdout is something like this:
- # b'Linux vyos 4.19.101-amd64-vyos #1 SMP Sun Feb 2 10:18:07 UTC 2020 x86_64 GNU/Linux\n'
+ # b'Linux LR1.wue3 5.10.61-amd64-vyos #1 SMP Fri Aug 27 08:55:46 UTC 2021 x86_64 GNU/Linux\n'
self.assertTrue(len(stdout) > 40)
+ def test_system_user_ssh_key(self):
+ ssh_user = 'ssh-test_user'
+ public_keys = 'vyos'
+ type = 'ssh-rsa'
+
+ self.cli_set(base_path + ['user', ssh_user, 'authentication', 'public-keys', public_keys, 'key', ssh_pubkey.replace('\n','')])
+
+ # check validate() - missing type for public-key
+ with self.assertRaises(ConfigSessionError):
+ self.cli_commit()
+ self.cli_set(base_path + ['user', ssh_user, 'authentication', 'public-keys', public_keys, 'type', type])
+
+ self.cli_commit()
+
+ # Check that SSH key was written properly
+ tmp = cmd(f'sudo cat /home/{ssh_user}/.ssh/authorized_keys')
+ key = f'{type} ' + ssh_pubkey.replace('\n','')
+ self.assertIn(key, tmp)
+
+ self.cli_delete(base_path + ['user', ssh_user])
+
def test_radius_kernel_features(self):
# T2886: RADIUS requires some Kernel options to be present
kernel = platform.release()
kernel_config = read_file(f'/boot/config-{kernel}')
# T2886 - RADIUS authentication - check for statically compiled options
options = ['CONFIG_AUDIT', 'CONFIG_AUDITSYSCALL', 'CONFIG_AUDIT_ARCH']
if LooseVersion(kernel_version()) < LooseVersion('5.0'):
options.append('CONFIG_AUDIT_WATCH')
options.append('CONFIG_AUDIT_TREE')
for option in options:
self.assertIn(f'{option}=y', kernel_config)
def test_system_login_radius_ipv4(self):
# Verify generated RADIUS configuration files
radius_key = 'VyOSsecretVyOS'
radius_server = '172.16.100.10'
radius_source = '127.0.0.1'
radius_port = '2000'
radius_timeout = '1'
self.cli_set(base_path + ['radius', 'server', radius_server, 'key', radius_key])
self.cli_set(base_path + ['radius', 'server', radius_server, 'port', radius_port])
self.cli_set(base_path + ['radius', 'server', radius_server, 'timeout', radius_timeout])
self.cli_set(base_path + ['radius', 'source-address', radius_source])
self.cli_set(base_path + ['radius', 'source-address', inc_ip(radius_source, 1)])
# check validate() - Only one IPv4 source-address supported
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_delete(base_path + ['radius', 'source-address', inc_ip(radius_source, 1)])
self.cli_commit()
# this file must be read with higher permissions
pam_radius_auth_conf = cmd('sudo cat /etc/pam_radius_auth.conf')
tmp = re.findall(r'\n?{}:{}\s+{}\s+{}\s+{}'.format(radius_server,
radius_port, radius_key, radius_timeout,
radius_source), pam_radius_auth_conf)
self.assertTrue(tmp)
# required, static options
self.assertIn('priv-lvl 15', pam_radius_auth_conf)
self.assertIn('mapped_priv_user radius_priv_user', pam_radius_auth_conf)
# PAM
pam_common_account = read_file('/etc/pam.d/common-account')
self.assertIn('pam_radius_auth.so', pam_common_account)
pam_common_auth = read_file('/etc/pam.d/common-auth')
self.assertIn('pam_radius_auth.so', pam_common_auth)
pam_common_session = read_file('/etc/pam.d/common-session')
self.assertIn('pam_radius_auth.so', pam_common_session)
pam_common_session_noninteractive = read_file('/etc/pam.d/common-session-noninteractive')
self.assertIn('pam_radius_auth.so', pam_common_session_noninteractive)
# NSS
nsswitch_conf = read_file('/etc/nsswitch.conf')
tmp = re.findall(r'passwd:\s+mapuid\s+files\s+mapname', nsswitch_conf)
self.assertTrue(tmp)
tmp = re.findall(r'group:\s+mapname\s+files', nsswitch_conf)
self.assertTrue(tmp)
def test_system_login_radius_ipv6(self):
# Verify generated RADIUS configuration files
radius_key = 'VyOS-VyOS'
radius_server = '2001:db8::1'
radius_source = '::1'
radius_port = '4000'
radius_timeout = '4'
self.cli_set(base_path + ['radius', 'server', radius_server, 'key', radius_key])
self.cli_set(base_path + ['radius', 'server', radius_server, 'port', radius_port])
self.cli_set(base_path + ['radius', 'server', radius_server, 'timeout', radius_timeout])
self.cli_set(base_path + ['radius', 'source-address', radius_source])
self.cli_set(base_path + ['radius', 'source-address', inc_ip(radius_source, 1)])
# check validate() - Only one IPv4 source-address supported
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_delete(base_path + ['radius', 'source-address', inc_ip(radius_source, 1)])
self.cli_commit()
# this file must be read with higher permissions
pam_radius_auth_conf = cmd('sudo cat /etc/pam_radius_auth.conf')
tmp = re.findall(r'\n?\[{}\]:{}\s+{}\s+{}\s+\[{}\]'.format(radius_server,
radius_port, radius_key, radius_timeout,
radius_source), pam_radius_auth_conf)
self.assertTrue(tmp)
# required, static options
self.assertIn('priv-lvl 15', pam_radius_auth_conf)
self.assertIn('mapped_priv_user radius_priv_user', pam_radius_auth_conf)
# PAM
pam_common_account = read_file('/etc/pam.d/common-account')
self.assertIn('pam_radius_auth.so', pam_common_account)
pam_common_auth = read_file('/etc/pam.d/common-auth')
self.assertIn('pam_radius_auth.so', pam_common_auth)
pam_common_session = read_file('/etc/pam.d/common-session')
self.assertIn('pam_radius_auth.so', pam_common_session)
pam_common_session_noninteractive = read_file('/etc/pam.d/common-session-noninteractive')
self.assertIn('pam_radius_auth.so', pam_common_session_noninteractive)
# NSS
nsswitch_conf = read_file('/etc/nsswitch.conf')
tmp = re.findall(r'passwd:\s+mapuid\s+files\s+mapname', nsswitch_conf)
self.assertTrue(tmp)
tmp = re.findall(r'group:\s+mapname\s+files', nsswitch_conf)
self.assertTrue(tmp)
if __name__ == '__main__':
- unittest.main(verbosity=2)
+ unittest.main(verbosity=2, failfast=True)
diff --git a/src/conf_mode/system-login.py b/src/conf_mode/system-login.py
index 59ea1d34b..78830931d 100755
--- a/src/conf_mode/system-login.py
+++ b/src/conf_mode/system-login.py
@@ -1,312 +1,312 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-2021 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# 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 os
from crypt import crypt
from crypt import METHOD_SHA512
from psutil import users
from pwd import getpwall
from pwd import getpwnam
from spwd import getspnam
from sys import exit
from vyos.config import Config
from vyos.configdict import dict_merge
from vyos.configverify import verify_vrf
from vyos.template import render
from vyos.template import is_ipv4
from vyos.util import cmd
from vyos.util import call
from vyos.util import DEVNULL
from vyos.util import dict_search
from vyos.xml import defaults
from vyos import ConfigError
from vyos import airbag
airbag.enable()
radius_config_file = "/etc/pam_radius_auth.conf"
def get_local_users():
"""Return list of dynamically allocated users (see Debian Policy Manual)"""
local_users = []
for s_user in getpwall():
uid = getpwnam(s_user.pw_name).pw_uid
if uid in range(1000, 29999):
if s_user.pw_name not in ['radius_user', 'radius_priv_user']:
local_users.append(s_user.pw_name)
return local_users
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['system', 'login']
login = conf.get_config_dict(base, key_mangling=('-', '_'),
- get_first_key=True)
+ no_tag_node_value_mangle=True, get_first_key=True)
# users no longer existing in the running configuration need to be deleted
local_users = get_local_users()
cli_users = []
if 'user' in login:
cli_users = list(login['user'])
# XXX: T2665: we can not safely rely on the defaults() when there are
# tagNodes in place, it is better to blend in the defaults manually.
default_values = defaults(base + ['user'])
for user in login['user']:
login['user'][user] = dict_merge(default_values, login['user'][user])
# XXX: T2665: we can not safely rely on the defaults() when there are
# tagNodes in place, it is better to blend in the defaults manually.
default_values = defaults(base + ['radius', 'server'])
for server in dict_search('radius.server', login) or []:
login['radius']['server'][server] = dict_merge(default_values,
login['radius']['server'][server])
# XXX: for a yet unknown reason when we only have one source-address
# get_config_dict() will show a string over a string
if 'radius' in login and 'source_address' in login['radius']:
if isinstance(login['radius']['source_address'], str):
login['radius']['source_address'] = [login['radius']['source_address']]
# create a list of all users, cli and users
all_users = list(set(local_users + cli_users))
# We will remove any normal users that dos not exist in the current
# configuration. This can happen if user is added but configuration was not
# saved and system is rebooted.
rm_users = [tmp for tmp in all_users if tmp not in cli_users]
if rm_users: login.update({'rm_users' : rm_users})
return login
def verify(login):
if 'rm_users' in login:
cur_user = os.environ['SUDO_USER']
if cur_user in login['rm_users']:
raise ConfigError(f'Attempting to delete current user: {cur_user}')
if 'user' in login:
system_users = getpwall()
for user, user_config in login['user'].items():
# Linux system users range up until UID 1000, we can not create a
# VyOS CLI user which already exists as system user
for s_user in system_users:
if s_user.pw_name == user and s_user.pw_uid < 1000:
raise ConfigError(f'User "{user}" can not be created, conflict with local system account!')
for pubkey, pubkey_options in (dict_search('authentication.public_keys', user_config) or {}).items():
if 'type' not in pubkey_options:
raise ConfigError(f'Missing type for public-key "{pubkey}"!')
if 'key' not in pubkey_options:
raise ConfigError(f'Missing key for public-key "{pubkey}"!')
# At lease one RADIUS server must not be disabled
if 'radius' in login:
if 'server' not in login['radius']:
raise ConfigError('No RADIUS server defined!')
fail = True
for server, server_config in dict_search('radius.server', login).items():
if 'key' not in server_config:
raise ConfigError(f'RADIUS server "{server}" requires key!')
if 'disabled' not in server_config:
fail = False
continue
if fail:
raise ConfigError('All RADIUS servers are disabled')
verify_vrf(login['radius'])
if 'source_address' in login['radius']:
ipv4_count = 0
ipv6_count = 0
for address in login['radius']['source_address']:
if is_ipv4(address): ipv4_count += 1
else: ipv6_count += 1
if ipv4_count > 1:
raise ConfigError('Only one IPv4 source-address can be set!')
if ipv6_count > 1:
raise ConfigError('Only one IPv6 source-address can be set!')
return None
def generate(login):
# calculate users encrypted password
if 'user' in login:
for user, user_config in login['user'].items():
tmp = dict_search('authentication.plaintext_password', user_config)
if tmp:
encrypted_password = crypt(tmp, METHOD_SHA512)
login['user'][user]['authentication']['encrypted_password'] = encrypted_password
del login['user'][user]['authentication']['plaintext_password']
# remove old plaintext password and set new encrypted password
env = os.environ.copy()
env['vyos_libexec_dir'] = '/usr/libexec/vyos'
# Set default commands for re-adding user with encrypted password
del_user_plain = f"system login user '{user}' authentication plaintext-password"
add_user_encrypt = f"system login user '{user}' authentication encrypted-password '{encrypted_password}'"
lvl = env['VYATTA_EDIT_LEVEL']
# We're in config edit level, for example "edit system login"
# Change default commands for re-adding user with encrypted password
if lvl != '/':
# Replace '/system/login' to 'system login'
lvl = lvl.strip('/').split('/')
# Convert command str to list
del_user_plain = del_user_plain.split()
# New command exclude level, for example "edit system login"
del_user_plain = del_user_plain[len(lvl):]
# Convert string to list
del_user_plain = " ".join(del_user_plain)
add_user_encrypt = add_user_encrypt.split()
add_user_encrypt = add_user_encrypt[len(lvl):]
add_user_encrypt = " ".join(add_user_encrypt)
call(f"/opt/vyatta/sbin/my_delete {del_user_plain}", env=env)
call(f"/opt/vyatta/sbin/my_set {add_user_encrypt}", env=env)
else:
try:
if getspnam(user).sp_pwdp == dict_search('authentication.encrypted_password', user_config):
# If the current encrypted bassword matches the encrypted password
# from the config - do not update it. This will remove the encrypted
# value from the system logs.
#
# The encrypted password will be set only once during the first boot
# after an image upgrade.
del login['user'][user]['authentication']['encrypted_password']
except:
pass
if 'radius' in login:
render(radius_config_file, 'login/pam_radius_auth.conf.tmpl', login,
permission=0o600, user='root', group='root')
else:
if os.path.isfile(radius_config_file):
os.unlink(radius_config_file)
return None
def apply(login):
if 'user' in login:
for user, user_config in login['user'].items():
# make new user using vyatta shell and make home directory (-m),
# default group of 100 (users)
command = 'useradd -m -N'
# check if user already exists:
if user in get_local_users():
# update existing account
command = 'usermod'
# all accounts use /bin/vbash
command += ' -s /bin/vbash'
# we need to use '' quotes when passing formatted data to the shell
# else it will not work as some data parts are lost in translation
tmp = dict_search('authentication.encrypted_password', user_config)
if tmp: command += f" -p '{tmp}'"
tmp = dict_search('full_name', user_config)
if tmp: command += f" -c '{tmp}'"
tmp = dict_search('home_directory', user_config)
if tmp: command += f" -d '{tmp}'"
else: command += f" -d '/home/{user}'"
command += f' -G frrvty,vyattacfg,sudo,adm,dip,disk {user}'
try:
cmd(command)
# we should not rely on the value stored in
# user_config['home_directory'], as a crazy user will choose
# username root or any other system user which will fail.
#
# XXX: Should we deny using root at all?
home_dir = getpwnam(user).pw_dir
render(f'{home_dir}/.ssh/authorized_keys', 'login/authorized_keys.tmpl',
user_config, permission=0o600, user=user, group='users')
except Exception as e:
raise ConfigError(f'Adding user "{user}" raised exception: "{e}"')
if 'rm_users' in login:
for user in login['rm_users']:
try:
# Logout user if he is still logged in
if user in list(set([tmp[0] for tmp in users()])):
print(f'{user} is logged in, forcing logout!')
call(f'pkill -HUP -u {user}')
# Remove user account but leave home directory to be safe
call(f'userdel -r {user}', stderr=DEVNULL)
except Exception as e:
raise ConfigError(f'Deleting user "{user}" raised exception: {e}')
#
# RADIUS configuration
#
env = os.environ.copy()
env['DEBIAN_FRONTEND'] = 'noninteractive'
try:
if 'radius' in login:
# Enable RADIUS in PAM
cmd('pam-auth-update --package --enable radius', env=env)
# Make NSS system aware of RADIUS
# This fancy snipped was copied from old Vyatta code
command = "sed -i -e \'/\smapname/b\' \
-e \'/^passwd:/s/\s\s*/&mapuid /\' \
-e \'/^passwd:.*#/s/#.*/mapname &/\' \
-e \'/^passwd:[^#]*$/s/$/ mapname &/\' \
-e \'/^group:.*#/s/#.*/ mapname &/\' \
-e \'/^group:[^#]*$/s/: */&mapname /\' \
/etc/nsswitch.conf"
else:
# Disable RADIUS in PAM
cmd('pam-auth-update --package --remove radius', env=env)
# Drop RADIUS from NSS NSS system
# This fancy snipped was copied from old Vyatta code
command = "sed -i -e \'/^passwd:.*mapuid[ \t]/s/mapuid[ \t]//\' \
-e \'/^passwd:.*[ \t]mapname/s/[ \t]mapname//\' \
-e \'/^group:.*[ \t]mapname/s/[ \t]mapname//\' \
-e \'s/[ \t]*$//\' \
/etc/nsswitch.conf"
cmd(command)
except Exception as e:
raise ConfigError(f'RADIUS configuration failed: {e}')
return None
if __name__ == '__main__':
try:
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
exit(1)
File Metadata
Details
Attached
Mime Type
text/x-diff
Expires
Sat, Oct 17, 12:24 AM (8 h, 25 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
4416563
Default Alt Text
(23 KB)
Attached To
Mode
rVYOSONEX vyos-1x
Attached
Detach File
Event Timeline
Log In to Comment