- *Create User*::
- $ ceph dashboard ac-user-create [--force-password] <username> [<password>] [<rolename>] [<name>] [<email>] [--enabled]
+ $ ceph dashboard ac-user-create [--force-password] <username> [<password>] [<rolename>] [<name>] [<email>] [--enabled] [<pwd_expiration_date>]
- *Delete User*::
'token': JLeaf(str),
'username': JLeaf(str),
'permissions': JObj(sub_elems={}, allow_unknown=True),
- 'sso': JLeaf(bool)
+ 'sso': JLeaf(bool),
+ 'pwdExpirationDate': JLeaf(int, none=True)
}, allow_unknown=False))
self._validate_jwt_token(data['token'], "admin", data['permissions'])
import time
+from datetime import datetime, timedelta
+
from .helper import DashboardTestCase
class UserTest(DashboardTestCase):
@classmethod
- def _create_user(cls, username=None, password=None, name=None, email=None, roles=None, enabled=True):
+ def _create_user(cls, username=None, password=None, name=None, email=None, roles=None,
+ enabled=True, pwd_expiration_date=None):
data = {}
if username:
data['username'] = username
data['email'] = email
if roles:
data['roles'] = roles
+ if pwd_expiration_date:
+ data['pwdExpirationDate'] = pwd_expiration_date
data['enabled'] = enabled
cls._post("/api/user", data)
@classmethod
- def _reset_login_to_admin(cls, username):
+ def _reset_login_to_admin(cls, username=None):
cls.logout()
- cls.delete_user(username)
+ if username:
+ cls.delete_user(username)
cls.login('admin', 'admin')
def test_crud_user(self):
'email': 'my@email.com',
'roles': ['administrator'],
'lastUpdate': user['lastUpdate'],
- 'enabled': True
+ 'enabled': True,
+ 'pwdExpirationDate': None
})
self._put('/api/user/user1', {
'email': 'mynew@email.com',
'roles': ['block-manager'],
'lastUpdate': user['lastUpdate'],
- 'enabled': True
+ 'enabled': True,
+ 'pwdExpirationDate': None
})
self._delete('/api/user/user1')
'email': 'klara@musterfrau.com',
'roles': ['administrator'],
'lastUpdate': user['lastUpdate'],
- 'enabled': False
+ 'enabled': False,
+ 'pwdExpirationDate': None
})
self._delete('/api/user/klara')
'email': None,
'roles': ['administrator'],
'lastUpdate': user['lastUpdate'],
- 'enabled': True
+ 'enabled': True,
+ 'pwdExpirationDate': None
}])
def test_create_user_already_exists(self):
'test4', 'bar'])
self.assertNotEqual(exitcode, 0)
self.delete_user('test4')
+
+ def test_create_user_with_pwd_expiration_date(self):
+ future_date = datetime.utcnow() + timedelta(days=10)
+ future_date = int(time.mktime(future_date.timetuple()))
+
+ self._create_user(username='user1',
+ password='mypassword10#',
+ name='My Name',
+ email='my@email.com',
+ roles=['administrator'],
+ pwd_expiration_date=future_date)
+ self.assertStatus(201)
+ user = self.jsonBody()
+
+ self._get('/api/user/user1')
+ self.assertStatus(200)
+ self.assertJsonBody({
+ 'username': 'user1',
+ 'name': 'My Name',
+ 'email': 'my@email.com',
+ 'roles': ['administrator'],
+ 'lastUpdate': user['lastUpdate'],
+ 'enabled': True,
+ 'pwdExpirationDate': future_date
+ })
+ self._delete('/api/user/user1')
+
+ def test_create_with_pwd_expiration_date_not_valid(self):
+ past_date = datetime.utcnow() - timedelta(days=10)
+ past_date = int(time.mktime(past_date.timetuple()))
+
+ self._create_user(username='user1',
+ password='mypassword10#',
+ name='My Name',
+ email='my@email.com',
+ roles=['administrator'],
+ pwd_expiration_date=past_date)
+ self.assertStatus(400)
+ self.assertError(code='pwd_past_expiration_date', component='user')
+
+ def test_create_with_default_expiration_date(self):
+ future_date_1 = datetime.utcnow() + timedelta(days=10)
+ future_date_1 = int(time.mktime(future_date_1.timetuple()))
+ future_date_2 = datetime.utcnow() + timedelta(days=11)
+ future_date_2 = int(time.mktime(future_date_2.timetuple()))
+
+ self._ceph_cmd(['dashboard', 'set-user-pwd-expiration-span', '10'])
+ self._create_user(username='user1',
+ password='mypassword10#',
+ name='My Name',
+ email='my@email.com',
+ roles=['administrator'])
+ self.assertStatus(201)
+
+ user = self._get('/api/user/user1')
+ self.assertStatus(200)
+ self.assertIsNotNone(user['pwdExpirationDate'])
+ self.assertGreater(user['pwdExpirationDate'], future_date_1)
+ self.assertLess(user['pwdExpirationDate'], future_date_2)
+
+ self._delete('/api/user/user1')
+ self._ceph_cmd(['dashboard', 'set-user-pwd-expiration-span', '0'])
+
+ def test_pwd_expiration_date_update(self):
+ self._ceph_cmd(['dashboard', 'set-user-pwd-expiration-span', '10'])
+ self._create_user(username='user1',
+ password='mypassword10#',
+ name='My Name',
+ email='my@email.com',
+ roles=['administrator'])
+ self.assertStatus(201)
+
+ user_1 = self._get('/api/user/user1')
+ self.assertStatus(200)
+
+ self.login('user1', 'mypassword10#')
+ self._post('/api/user/user1/change_password', {
+ 'old_password': 'mypassword10#',
+ 'new_password': 'newpassword01#'
+ })
+ self.assertStatus(200)
+ self._reset_login_to_admin()
+
+ user_2 = self._get('/api/user/user1')
+ self.assertStatus(200)
+ self.assertLess(user_1['pwdExpirationDate'], user_2['pwdExpirationDate'])
+
+ self._delete('/api/user/user1')
+ self._ceph_cmd(['dashboard', 'set-user-pwd-expiration-span', '0'])
max-args=5
# Maximum number of attributes for a class (see R0902).
-max-attributes=7
+max-attributes=10
# Maximum number of boolean expressions in a if statement
max-bool-expr=5
"""
def create(self, username, password):
- user_perms = AuthManager.authenticate(username, password)
+ user_data = AuthManager.authenticate(username, password)
+ user_perms, pwd_expiration_date = None, None
+ if user_data:
+ user_perms = user_data.get('permissions')
+ pwd_expiration_date = user_data.get('pwdExpirationDate')
+
if user_perms is not None:
logger.debug('Login successful')
token = JwtManager.gen_token(username)
'token': token,
'username': username,
'permissions': user_perms,
+ 'pwdExpirationDate': pwd_expiration_date,
'sso': mgr.SSO_DB.protocol == 'saml2'
}
import cherrypy
-from . import ApiController, RESTController
+from . import ApiController, RESTController, UiApiController
from ..settings import Settings as SettingsModule, Options
from ..security import Scope
with self._attribute_handler(kwargs) as data:
for name, value in data.items():
setattr(SettingsModule, self._to_native(name), value)
+
+
+@UiApiController('/standard_settings')
+class StandardSettings(RESTController):
+ def list(self):
+ return {
+ 'user_pwd_expiration_span': SettingsModule.USER_PWD_EXPIRATION_SPAN,
+ 'user_pwd_expiration_warning_1': SettingsModule.USER_PWD_EXPIRATION_WARNING_1,
+ 'user_pwd_expiration_warning_2': SettingsModule.USER_PWD_EXPIRATION_WARNING_2
+ }
# -*- coding: utf-8 -*-
from __future__ import absolute_import
+from datetime import datetime
+
+import time
+
import cherrypy
from . import BaseController, ApiController, RESTController, Endpoint
from .. import mgr
from ..exceptions import DashboardException, UserAlreadyExists, \
- UserDoesNotExist, PasswordCheckException
+ UserDoesNotExist, PasswordCheckException, PwdExpirationDateNotValid
from ..security import Scope
from ..services.access_control import SYSTEM_ROLES, PasswordCheck
from ..services.auth import JwtManager
return User._user_to_dict(user)
def create(self, username=None, password=None, name=None, email=None,
- roles=None, enabled=True):
+ roles=None, enabled=True, pwdExpirationDate=None):
if not username:
raise DashboardException(msg='Username is required',
code='username_required',
validate_password_policy(password, username)
try:
user = mgr.ACCESS_CTRL_DB.create_user(username, password, name,
- email, enabled)
+ email, enabled, pwdExpirationDate)
except UserAlreadyExists:
raise DashboardException(msg='Username already exists',
code='username_already_exists',
component='user')
+ except PwdExpirationDateNotValid:
+ raise DashboardException(msg='Password expiration date must not be in '
+ 'the past',
+ code='pwd_past_expiration_date',
+ component='user')
+
if user_roles:
user.set_roles(user_roles)
mgr.ACCESS_CTRL_DB.save()
mgr.ACCESS_CTRL_DB.save()
def set(self, username, password=None, name=None, email=None, roles=None,
- enabled=None):
+ enabled=None, pwdExpirationDate=None):
if JwtManager.get_username() == username and enabled is False:
raise DashboardException(msg='You are not allowed to disable your user',
code='cannot_disable_current_user',
if password:
validate_password_policy(password, username)
user.set_password(password)
+ if pwdExpirationDate and \
+ (pwdExpirationDate < int(time.mktime(datetime.utcnow().timetuple()))):
+ raise DashboardException(
+ msg='Password expiration date must not be in the past',
+ code='pwd_past_expiration_date', component='user')
user.name = name
user.email = email
if enabled is not None:
user.enabled = enabled
+ user.pwd_expiration_date = pwdExpirationDate
user.set_roles(user_roles)
mgr.ACCESS_CTRL_DB.save()
return User._user_to_dict(user)
.format(rolename, username))
+class PwdExpirationDateNotValid(Exception):
+ def __init__(self):
+ super(PwdExpirationDateNotValid, self).__init__(
+ "The password expiration date must not be in the past")
+
+
class GrafanaError(Exception):
pass
import { NgBootstrapFormValidationModule } from 'ng-bootstrap-form-validation';
import { ButtonsModule } from 'ngx-bootstrap/buttons';
+import { BsDatepickerModule } from 'ngx-bootstrap/datepicker';
import { BsDropdownModule } from 'ngx-bootstrap/dropdown';
import { PopoverModule } from 'ngx-bootstrap/popover';
import { TabsModule } from 'ngx-bootstrap/tabs';
SharedModule,
TabsModule.forRoot(),
RouterModule,
- NgBootstrapFormValidationModule
+ NgBootstrapFormValidationModule,
+ BsDatepickerModule.forRoot()
],
declarations: [
LoginComponent,
window.location.replace(login.login_url);
}
} else {
- this.authStorageService.set(login.username, token, login.permissions, login.sso);
+ this.authStorageService.set(
+ login.username,
+ token,
+ login.permissions,
+ login.sso,
+ login.pwdExpirationDate
+ );
this.router.navigate(['']);
}
});
+<cd-loading-panel *ngIf="!pwdExpirationSettings"
+ i18n>Loading...</cd-loading-panel>
+
<div class="col-sm-12 col-lg-6">
<form name="userForm"
#formDir="ngForm"
[formGroup]="userForm"
+ *ngIf="pwdExpirationSettings"
novalidate>
<div class="card">
<div i18n="form title|Example: Create Pool@@formTitle"
</div>
</div>
+ <!-- Password expiration date -->
+ <div class="form-group row"
+ *ngIf="showExpirationDateField()">
+ <label class="col-form-label col-sm-3"
+ for="pwdExpirationDate">
+ <ng-container i18n>Password expiration date</ng-container>
+ <span class="required"
+ *ngIf="pwdExpirationSettings.pwdExpirationSpan > 0"></span>
+ </label>
+ <div class="col-sm-9">
+ <div class="input-group">
+ <input type="text"
+ class="form-control"
+ i18n-placeholder
+ placeholder="Password expiration date..."
+ [bsConfig]="bsConfig"
+ [minDate]="minDate"
+ bsDatepicker
+ id="pwdExpirationDate"
+ name="pwdExpirationDate"
+ formControlName="pwdExpirationDate">
+ <span class="input-group-append">
+ <button type="button"
+ class="btn btn-light"
+ (click)="clearExpirationDate()">
+ <i class="icon-prepend {{ icons.destroy }}"></i>
+ </button>
+ </span>
+ <span class="invalid-feedback"
+ *ngIf="userForm.showError('pwdExpirationDate', formDir, 'required')"
+ i18n>This field is required.</span>
+ </div>
+ </div>
+ </div>
+
<!-- Name -->
<div class="form-group row">
<label i18n
import { RouterTestingModule } from '@angular/router/testing';
import { ButtonsModule } from 'ngx-bootstrap/buttons';
+import { BsDatepickerModule } from 'ngx-bootstrap/datepicker';
import { BsModalService } from 'ngx-bootstrap/modal';
import { ToastrModule } from 'ngx-toastr';
import { of } from 'rxjs';
import { configureTestBed, FormHelper, i18nProviders } from '../../../../testing/unit-test-helper';
import { RoleService } from '../../../shared/api/role.service';
+import { SettingsService } from '../../../shared/api/settings.service';
import { UserService } from '../../../shared/api/user.service';
import { ComponentsModule } from '../../../shared/components/components.module';
import { CdFormGroup } from '../../../shared/forms/cd-form-group';
ComponentsModule,
ToastrModule.forRoot(),
SharedModule,
- ButtonsModule.forRoot()
+ ButtonsModule.forRoot(),
+ BsDatepickerModule.forRoot()
],
declarations: [UserFormComponent, FakeComponent],
providers: i18nProviders
});
it('should not disable fields', () => {
- ['username', 'name', 'password', 'confirmpassword', 'email', 'roles'].forEach((key) =>
- expect(form.get(key).disabled).toBeFalsy()
- );
+ [
+ 'username',
+ 'name',
+ 'password',
+ 'confirmpassword',
+ 'email',
+ 'roles',
+ 'pwdExpirationDate'
+ ].forEach((key) => expect(form.get(key).disabled).toBeFalsy());
});
it('should validate username required', () => {
name: 'User 0',
email: 'user0@email.com',
roles: ['administrator'],
- enabled: true
+ enabled: true,
+ pwdExpirationDate: undefined
};
formHelper.setMultipleValues(user);
formHelper.setValue('confirmpassword', user.password);
name: 'User 1',
email: 'user1@email.com',
roles: ['administrator'],
- enabled: true
+ enabled: true,
+ pwdExpirationDate: undefined
};
const roles = [
{
}
}
];
+ const pwdExpirationSettings = {
+ user_pwd_expiration_warning_1: 10,
+ user_pwd_expiration_warning_2: 5,
+ user_pwd_expiration_span: 90
+ };
beforeEach(() => {
spyOn(userService, 'get').and.callFake(() => of(user));
spyOn(TestBed.get(RoleService), 'list').and.callFake(() => of(roles));
setUrl('/user-management/users/edit/user1');
+ spyOn(TestBed.get(SettingsService), 'pwdExpirationSettings').and.callFake(() =>
+ of(pwdExpirationSettings)
+ );
component.ngOnInit();
const req = httpTesting.expectOne('api/role');
expect(req.request.method).toBe('GET');
req.flush(roles);
+ httpTesting.expectOne('ui-api/standard_settings');
});
afterEach(() => {
import { I18n } from '@ngx-translate/i18n-polyfill';
import * as _ from 'lodash';
import { BsModalRef, BsModalService } from 'ngx-bootstrap/modal';
+import { forkJoin as observableForkJoin } from 'rxjs';
import { AuthService } from '../../../shared/api/auth.service';
import { RoleService } from '../../../shared/api/role.service';
+import { SettingsService } from '../../../shared/api/settings.service';
import { UserService } from '../../../shared/api/user.service';
import { ConfirmationModalComponent } from '../../../shared/components/confirmation-modal/confirmation-modal.component';
import { SelectMessages } from '../../../shared/components/select/select-messages.model';
import { NotificationType } from '../../../shared/enum/notification-type.enum';
import { CdFormGroup } from '../../../shared/forms/cd-form-group';
import { CdValidators } from '../../../shared/forms/cd-validators';
+import { CdPwdExpirationSettings } from '../../../shared/models/cd-pwd-expiration-settings';
import { AuthStorageService } from '../../../shared/services/auth-storage.service';
import { NotificationService } from '../../../shared/services/notification.service';
import { UserChangePasswordService } from '../../../shared/services/user-change-password.service';
passwordStrengthLevel: string;
passwordStrengthDescription: string;
icons = Icons;
+ minDate: Date;
+ bsConfig = {
+ dateInputFormat: 'YYYY-MM-DD',
+ containerClass: 'theme-default'
+ };
+ pwdExpirationSettings: CdPwdExpirationSettings;
constructor(
private authService: AuthService,
private notificationService: NotificationService,
private i18n: I18n,
public actionLabels: ActionLabelsI18n,
- private userChangePasswordService: UserChangePasswordService
+ private userChangePasswordService: UserChangePasswordService,
+ private settingsService: SettingsService
) {
this.resource = this.i18n('user');
this.createForm();
updateOn: 'blur',
validators: []
}),
+ pwdExpirationDate: new FormControl(''),
email: new FormControl('', {
validators: [Validators.email]
}),
} else {
this.action = this.actionLabels.CREATE;
}
+ this.minDate = new Date();
- this.roleService.list().subscribe((roles: Array<UserFormRoleModel>) => {
- this.allRoles = _.map(roles, (role) => {
- role.enabled = true;
- return role;
- });
- });
- if (this.mode === this.userFormMode.editing) {
- this.initEdit();
- }
+ const observables = [this.roleService.list(), this.settingsService.pwdExpirationSettings()];
+ observableForkJoin(observables).subscribe(
+ (result: [UserFormRoleModel[], CdPwdExpirationSettings]) => {
+ this.allRoles = _.map(result[0], (role) => {
+ role.enabled = true;
+ return role;
+ });
+ this.pwdExpirationSettings = new CdPwdExpirationSettings(result[1]);
+
+ if (this.mode === this.userFormMode.editing) {
+ this.initEdit();
+ } else {
+ if (this.pwdExpirationSettings.pwdExpirationSpan > 0) {
+ const pwdExpirationDateField = this.userForm.get('pwdExpirationDate');
+ const expirationDate = new Date();
+ expirationDate.setDate(
+ this.minDate.getDate() + this.pwdExpirationSettings.pwdExpirationSpan
+ );
+ pwdExpirationDateField.setValue(expirationDate);
+ pwdExpirationDateField.setValidators([Validators.required]);
+ }
+ }
+ }
+ );
}
initEdit() {
['username', 'name', 'email', 'roles', 'enabled'].forEach((key) =>
this.userForm.get(key).setValue(response[key])
);
+ const expirationDate = response['pwdExpirationDate'];
+ if (expirationDate) {
+ this.userForm.get('pwdExpirationDate').setValue(new Date(expirationDate * 1000));
+ }
}
getRequest(): UserFormModel {
['username', 'password', 'name', 'email', 'roles', 'enabled'].forEach(
(key) => (userFormModel[key] = this.userForm.get(key).value)
);
+ const expirationDate = this.userForm.get('pwdExpirationDate').value;
+ if (expirationDate) {
+ if (
+ this.mode !== this.userFormMode.editing ||
+ this.response.pwdExpirationDate !== Number(expirationDate) / 1000
+ ) {
+ expirationDate.setHours(23, 59, 59);
+ }
+ userFormModel['pwdExpirationDate'] = Number(expirationDate) / 1000;
+ }
return userFormModel;
}
return password && this.passwordStrengthLevel === 'passwordStrengthLevel0';
}
+ showExpirationDateField() {
+ return (
+ this.userForm.getValue('pwdExpirationDate') > 0 ||
+ this.userForm.touched ||
+ this.pwdExpirationSettings.pwdExpirationSpan > 0
+ );
+ }
+
public isCurrentUser(): boolean {
return this.authStorageService.getUsername() === this.userForm.getValue('username');
}
);
}
+ clearExpirationDate() {
+ this.userForm.get('pwdExpirationDate').setValue(undefined);
+ }
+
submit() {
if (this.mode === this.userFormMode.editing) {
this.editAction();
export class UserFormModel {
username: string;
password: string;
+ pwdExpirationDate: number;
name: string;
email: string;
roles: Array<string>;
import { CdTableColumn } from '../../../shared/models/cd-table-column';
import { CdTableSelection } from '../../../shared/models/cd-table-selection';
import { Permission } from '../../../shared/models/permissions';
+import { CdDatePipe } from '../../../shared/pipes/cd-date.pipe';
import { EmptyPipe } from '../../../shared/pipes/empty.pipe';
import { AuthStorageService } from '../../../shared/services/auth-storage.service';
import { NotificationService } from '../../../shared/services/notification.service';
private authStorageService: AuthStorageService,
private i18n: I18n,
private urlBuilder: URLBuilderService,
+ private cdDatePipe: CdDatePipe,
public actionLabels: ActionLabelsI18n
) {
this.permission = this.authStorageService.getPermissions().user;
prop: 'enabled',
flexGrow: 1,
cellTransformation: CellTemplate.checkIcon
+ },
+ {
+ name: this.i18n('Password expiration date'),
+ prop: 'pwdExpirationDate',
+ flexGrow: 1,
+ pipe: this.cdDatePipe
}
];
}
getUsers() {
this.userService.list().subscribe((users: Array<any>) => {
+ users.forEach((user) => {
+ if (user['pwdExpirationDate'] && user['pwdExpirationDate'] > 0) {
+ user['pwdExpirationDate'] = user['pwdExpirationDate'] * 1000;
+ }
+ });
this.users = users;
});
}
login(credentials: Credentials): Observable<LoginResponse> {
return this.http.post('api/auth', credentials).pipe(
tap((resp: LoginResponse) => {
- this.authStorageService.set(resp.username, resp.token, resp.permissions, resp.sso);
+ this.authStorageService.set(
+ resp.username,
+ resp.token,
+ resp.permissions,
+ resp.sso,
+ resp.pwdExpirationDate
+ );
})
);
}
import { HttpClient } from '@angular/common/http';
import { Injectable } from '@angular/core';
-import * as _ from 'lodash';
+import { Observable } from 'rxjs';
+
+import { CdPwdExpirationSettings } from '../models/cd-pwd-expiration-settings';
import { ApiModule } from './api.module';
@Injectable({
validateGrafanaDashboardUrl(uid) {
return this.http.get(`api/grafana/validation/${uid}`);
}
+
+ pwdExpirationSettings(): Observable<CdPwdExpirationSettings> {
+ return this.http.get<CdPwdExpirationSettings>('ui-api/standard_settings');
+ }
}
--- /dev/null
+export class CdPwdExpirationSettings {
+ pwdExpirationSpan = 0;
+ pwdExpirationWarning1: number;
+ pwdExpirationWarning2: number;
+
+ constructor(data) {
+ this.pwdExpirationSpan = data.user_pwd_expiration_span;
+ this.pwdExpirationWarning1 = data.user_pwd_expiration_warning_1;
+ this.pwdExpirationWarning2 = data.user_pwd_expiration_warning_2;
+ }
+}
username: string;
token: string;
permissions: object;
+ pwdExpirationDate: number;
sso: boolean;
}
export class AuthStorageService {
constructor() {}
- set(username: string, token: string, permissions: object = {}, sso = false) {
+ set(
+ username: string,
+ token: string,
+ permissions: object = {},
+ sso = false,
+ pwdExpirationDate: number = null
+ ) {
localStorage.setItem('dashboard_username', username);
localStorage.setItem('access_token', token);
localStorage.setItem('dashboard_permissions', JSON.stringify(new Permissions(permissions)));
+ localStorage.setItem('user_pwd_expiration_date', String(pwdExpirationDate));
localStorage.setItem('sso', String(sso));
}
remove() {
localStorage.removeItem('access_token');
localStorage.removeItem('dashboard_username');
+ localStorage.removeItem('user_pwd_expiration_data');
}
getToken(): string {
);
}
+ getPwdExpirationDate(): number {
+ return Number(localStorage.getItem('user_pwd_expiration_date'));
+ }
+
isSSO() {
return localStorage.getItem('sso') === 'true';
}
import time
import re
+from datetime import datetime, timedelta
+
import bcrypt
from mgr_module import CLIReadCommand, CLIWriteCommand
from .. import mgr
from ..security import Scope, Permission
+from ..settings import Settings
from ..exceptions import RoleAlreadyExists, RoleDoesNotExist, ScopeNotValid, \
PermissionNotValid, RoleIsAssociatedWithUser, \
UserAlreadyExists, UserDoesNotExist, ScopeNotInRole, \
- RoleNotInUser, PasswordCheckException
+ RoleNotInUser, PasswordCheckException, PwdExpirationDateNotValid
logger = logging.getLogger('access_control')
class User(object):
def __init__(self, username, password, name=None, email=None, roles=None,
- last_update=None, enabled=True):
+ last_update=None, enabled=True, pwd_expiration_date=None):
self.username = username
self.password = password
self.name = name
else:
self.last_update = last_update
self._enabled = enabled
+ self.pwd_expiration_date = pwd_expiration_date
+ if self.pwd_expiration_date is None:
+ self.refresh_pwd_expiration_date()
def refresh_last_update(self):
self.last_update = int(time.time())
+ def refresh_pwd_expiration_date(self):
+ if Settings.USER_PWD_EXPIRATION_SPAN > 0:
+ expiration_date = datetime.utcnow() + timedelta(
+ days=Settings.USER_PWD_EXPIRATION_SPAN)
+ self.pwd_expiration_date = int(time.mktime(expiration_date.timetuple()))
+ else:
+ self.pwd_expiration_date = None
+
@property
def enabled(self):
return self._enabled
def set_password_hash(self, hashed_password):
self.password = hashed_password
self.refresh_last_update()
+ self.refresh_pwd_expiration_date()
def compare_password(self, password):
"""
pass_hash = password_hash(password, salt_password=self.password)
return pass_hash == self.password
+ def is_pwd_expired(self):
+ if self.pwd_expiration_date:
+ current_time = int(time.mktime(datetime.utcnow().timetuple()))
+ return self.pwd_expiration_date < current_time
+ return False
+
def set_roles(self, roles):
self.roles = set(roles)
self.refresh_last_update()
'name': self.name,
'email': self.email,
'lastUpdate': self.last_update,
- 'enabled': self.enabled
+ 'enabled': self.enabled,
+ 'pwdExpirationDate': self.pwd_expiration_date
}
@classmethod
def from_dict(cls, u_dict, roles):
return User(u_dict['username'], u_dict['password'], u_dict['name'],
u_dict['email'], {roles[r] for r in u_dict['roles']},
- u_dict['lastUpdate'], u_dict['enabled'])
+ u_dict['lastUpdate'], u_dict['enabled'], u_dict['pwdExpirationDate'])
class AccessControlDB(object):
del self.roles[name]
- def create_user(self, username, password, name, email, enabled=True):
+ def create_user(self, username, password, name, email, enabled=True, pwd_expiration_date=None):
logger.debug("creating user: username=%s", username)
with self.lock:
if username in self.users:
raise UserAlreadyExists(username)
- user = User(username, password_hash(password), name, email, enabled=enabled)
+ if pwd_expiration_date and \
+ (pwd_expiration_date < int(time.mktime(datetime.utcnow().timetuple()))):
+ raise PwdExpirationDateNotValid()
+ user = User(username, password_hash(password), name, email, enabled=enabled,
+ pwd_expiration_date=pwd_expiration_date)
self.users[username] = user
return user
for user, _ in v1_db['users'].items():
v1_db['users'][user]['enabled'] = True
+ v1_db['users'][user]['pwdExpirationDate'] = None
self.roles = {rn: Role.from_dict(r) for rn, r in v1_db.get('roles', {}).items()}
self.users = {un: User.from_dict(u, dict(self.roles, **SYSTEM_ROLES))
'name=name,type=CephString,req=false '
'name=email,type=CephString,req=false '
'name=enabled,type=CephBool,req=false '
- 'name=force_password,type=CephBool,req=false',
+ 'name=force_password,type=CephBool,req=false '
+ 'name=pwd_expiration_date,type=CephInt,req=false',
'Create a user')
def ac_user_create_cmd(_, username, password=None, rolename=None, name=None,
- email=None, enabled=True, force_password=False):
+ email=None, enabled=True, force_password=False,
+ pwd_expiration_date=None):
try:
role = mgr.ACCESS_CTRL_DB.get_role(rolename) if rolename else None
except RoleDoesNotExist as ex:
if not force_password:
pw_check = PasswordCheck(password, username)
pw_check.check_all()
- user = mgr.ACCESS_CTRL_DB.create_user(username, password, name, email, enabled)
+ user = mgr.ACCESS_CTRL_DB.create_user(username, password, name, email,
+ enabled, pwd_expiration_date)
except PasswordCheckException as ex:
return -errno.EINVAL, '', str(ex)
except UserAlreadyExists as ex:
try:
user = mgr.ACCESS_CTRL_DB.get_user(username)
if user.password:
- if user.enabled and user.compare_password(password):
- return user.permissions_dict()
+ if user.enabled and user.compare_password(password) \
+ and not user.is_pwd_expired():
+ return {'permissions': user.permissions_dict(),
+ 'pwdExpirationDate': user.pwd_expiration_date}
except UserDoesNotExist:
logger.debug("User '%s' does not exist", username)
return None
# iSCSI management settings
ISCSI_API_SSL_VERIFICATION = (True, bool)
+ # user management settings
+ # Time span of user passwords to expire in days.
+ # The default value is '0' which means that user passwords are
+ # never going to expire.
+ USER_PWD_EXPIRATION_SPAN = (0, int)
+ # warning levels to notify the user that the password is going
+ # to expire soon
+ USER_PWD_EXPIRATION_WARNING_1 = (10, int)
+ USER_PWD_EXPIRATION_WARNING_2 = (5, int)
+
@staticmethod
def has_default_value(name):
return getattr(Settings, name, None) is None or \
import time
import unittest
+from datetime import datetime, timedelta
+
from . import CmdException, CLICommandTestMixin
from .. import mgr
from ..security import Scope, Permission
def validate_persistent_user(self, username, roles, password=None,
name=None, email=None, last_update=None,
- enabled=True):
+ enabled=True, pwdExpirationDate=None):
db = self.load_persistent_db()
self.assertIn('users', db)
self.assertIn(username, db['users'])
self.assertEqual(db['users'][username]['email'], email)
if last_update:
self.assertEqual(db['users'][username]['lastUpdate'], last_update)
+ if pwdExpirationDate:
+ self.assertEqual(db['users'][username]['pwdExpirationDate'], pwdExpirationDate)
self.assertEqual(db['users'][username]['enabled'], enabled)
def validate_persistent_no_user(self, username):
self.assertEqual(str(ctx.exception),
"Cannot update system role 'read-only'")
- def test_create_user(self, username='admin', rolename=None, enabled=True):
+ def test_create_user(self, username='admin', rolename=None, enabled=True,
+ pwdExpirationDate=None):
user = self.exec_cmd('ac-user-create', username=username,
rolename=rolename, password='admin',
name='{} User'.format(username),
email='{}@user.com'.format(username),
- enabled=enabled, force_password=True)
+ enabled=enabled, force_password=True,
+ pwd_expiration_date=pwdExpirationDate)
pass_hash = password_hash('admin', user['password'])
self.assertDictEqual(user, {
'username': username,
'password': pass_hash,
+ 'pwdExpirationDate': pwdExpirationDate,
'lastUpdate': user['lastUpdate'],
'name': '{} User'.format(username),
'email': '{}@user.com'.format(username),
def test_create_disabled_user(self):
self.test_create_user(enabled=False)
+ def test_create_user_pwd_expiration_date(self):
+ expiration_date = datetime.utcnow() + timedelta(days=10)
+ expiration_date = int(time.mktime(expiration_date.timetuple()))
+ self.test_create_user(pwdExpirationDate=expiration_date)
+
def test_create_user_with_role(self):
self.test_add_role_scope_perms()
self.test_create_user(rolename='test_role')
'username': 'admin',
'lastUpdate': user['lastUpdate'],
'password': pass_hash,
+ 'pwdExpirationDate': None,
'name': 'admin User',
'email': 'admin@user.com',
'roles': ['block-manager', 'pool-manager'],
self.assertDictEqual(user, {
'username': 'admin',
'password': pass_hash,
+ 'pwdExpirationDate': None,
'name': 'Admin Name',
'email': 'admin@admin.com',
'lastUpdate': user['lastUpdate'],
self.assertDictEqual(user, {
'username': 'admin',
'password': pass_hash,
+ 'pwdExpirationDate': None,
'name': 'admin User',
'email': 'admin@user.com',
'lastUpdate': user['lastUpdate'],
self.assertDictEqual(user, {
'username': 'admin',
'password': pass_hash,
+ 'pwdExpirationDate': None,
'name': 'admin User',
'email': 'admin@user.com',
'lastUpdate': user['lastUpdate'],
self.assertDictEqual(user, {
'username': 'admin',
'password': pass_hash,
+ 'pwdExpirationDate': None,
'name': None,
'email': None,
'lastUpdate': user['lastUpdate'],
self.assertDictEqual(user, {
'username': 'admin',
'password': pass_hash,
+ 'pwdExpirationDate': None,
'name': 'admin User',
'email': 'admin@user.com',
'lastUpdate': user['lastUpdate'],
'lastUpdate': user['lastUpdate'],
'password':
"$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
+ 'pwdExpirationDate': None,
'name': 'admin User',
'email': 'admin@user.com',
'roles': ['block-manager', 'test_role'],
def test_load_v2(self):
"""
- The `enabled` attribute of a user has been added in v2
+ The `enabled` and `pwdExpirationDate` attributes of a user have been added in v2
"""
self.CONFIG_KEY_DICT['accessdb_v1'] = '''
{{
"username": "admin",
"password":
"$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
+ "pwdExpirationDate": null,
"roles": ["block-manager", "test_role"],
"name": "admin User",
"email": "admin@user.com",
'lastUpdate': user['lastUpdate'],
'password':
"$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
+ 'pwdExpirationDate': None,
'name': 'admin User',
'email': 'admin@user.com',
'roles': ['block-manager', 'test_role'],
'lastUpdate': user['lastUpdate'],
'password':
"$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
+ 'pwdExpirationDate': None,
'name': None,
'email': None,
'roles': ['administrator'],
projects / ceph.git / commitdiff