Source code for keystone.identity.backends.sql_model

# Copyright 2012 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import datetime

import sqlalchemy
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import orm

from keystone.common import sql
import keystone.conf


CONF = keystone.conf.CONF


[docs]class User(sql.ModelBase, sql.DictBase): __tablename__ = 'user' attributes = ['id', 'name', 'domain_id', 'password', 'enabled', 'default_project_id', 'password_expires_at'] readonly_attributes = ['id', 'password_expires_at'] id = sql.Column(sql.String(64), primary_key=True) _enabled = sql.Column('enabled', sql.Boolean) extra = sql.Column(sql.JsonBlob()) default_project_id = sql.Column(sql.String(64)) local_user = orm.relationship('LocalUser', uselist=False, single_parent=True, lazy='subquery', cascade='all,delete-orphan', backref='user') federated_users = orm.relationship('FederatedUser', single_parent=True, lazy='subquery', cascade='all,delete-orphan', backref='user') nonlocal_users = orm.relationship('NonLocalUser', single_parent=True, lazy='subquery', cascade='all,delete-orphan', backref='user') created_at = sql.Column(sql.DateTime, nullable=True) last_active_at = sql.Column(sql.Date, nullable=True) # name property @hybrid_property def name(self): if self.local_user: return self.local_user.name elif self.nonlocal_users: return self.nonlocal_users[0].name elif self.federated_users: return self.federated_users[0].display_name else: return None @name.setter def name(self, value): if not self.local_user: self.local_user = LocalUser() self.local_user.name = value @name.expression
[docs] def name(cls): return LocalUser.name # password properties
@hybrid_property def password_ref(self): """Return the current password.""" if self.local_user and self.local_user.passwords: return self.local_user.passwords[-1] return None @hybrid_property def password(self): if self.password_ref: return self.password_ref.password return None @hybrid_property def password_created_at(self): if self.password_ref: return self.password_ref.created_at return None @hybrid_property def password_expires_at(self): if self.password_ref: return self.password_ref.expires_at return None @hybrid_property def password_is_expired(self): if self.password_expires_at: return datetime.datetime.utcnow() >= self.password_expires_at return False @password.setter def password(self, value): now = datetime.datetime.utcnow() if not self.local_user: self.local_user = LocalUser() # truncate extra passwords if self.local_user.passwords: unique_cnt = CONF.security_compliance.unique_last_password_count self.local_user.passwords = self.local_user.passwords[-unique_cnt:] # set all previous passwords to be expired for ref in self.local_user.passwords: if not ref.expires_at or ref.expires_at > now: ref.expires_at = now new_password_ref = Password() new_password_ref.password = value new_password_ref.created_at = now new_password_ref.expires_at = self._get_password_expires_at(now) self.local_user.passwords.append(new_password_ref) def _get_password_expires_at(self, created_at): expires_days = CONF.security_compliance.password_expires_days ignore_list = CONF.security_compliance.password_expires_ignore_user_ids if expires_days and (self.id not in ignore_list): expired_date = (created_at + datetime.timedelta(days=expires_days)) return expired_date.replace(microsecond=0) return None @password.expression
[docs] def password(cls): return Password.password # domain_id property
@hybrid_property def domain_id(self): if self.local_user: return self.local_user.domain_id elif self.nonlocal_users: return self.nonlocal_users[0].domain_id else: return None @domain_id.setter def domain_id(self, value): if not self.local_user: self.local_user = LocalUser() self.local_user.domain_id = value @domain_id.expression
[docs] def domain_id(cls): return LocalUser.domain_id # enabled property
@hybrid_property def enabled(self): if self._enabled: max_days = ( CONF.security_compliance.disable_user_account_days_inactive) last_active = self.last_active_at if not last_active and self.created_at: last_active = self.created_at.date() if max_days and last_active: now = datetime.datetime.utcnow().date() days_inactive = (now - last_active).days if days_inactive >= max_days: self._enabled = False return self._enabled @enabled.setter def enabled(self, value): if (value and CONF.security_compliance.disable_user_account_days_inactive): self.last_active_at = datetime.datetime.utcnow().date() if value and self.local_user: self.local_user.failed_auth_count = 0 self.local_user.failed_auth_at = None self._enabled = value @enabled.expression
[docs] def enabled(cls): return User._enabled
[docs] def to_dict(self, include_extra_dict=False): d = super(User, self).to_dict(include_extra_dict=include_extra_dict) if 'default_project_id' in d and d['default_project_id'] is None: del d['default_project_id'] return d
@classmethod
[docs] def from_dict(cls, user_dict): """Override from_dict to remove password_expires_at attribute. Overriding this method to remove password_expires_at attribute to support update_user and unit tests where password_expires_at inadvertently gets added by calling to_dict followed by from_dict. :param user_dict: User entity dictionary :returns User: User object """ new_dict = user_dict.copy() password_expires_at_key = 'password_expires_at' if password_expires_at_key in user_dict: del new_dict[password_expires_at_key] return super(User, cls).from_dict(new_dict)
[docs]class LocalUser(sql.ModelBase, sql.DictBase): __tablename__ = 'local_user' attributes = ['id', 'user_id', 'domain_id', 'name'] id = sql.Column(sql.Integer, primary_key=True) user_id = sql.Column(sql.String(64), sql.ForeignKey('user.id', ondelete='CASCADE'), unique=True) domain_id = sql.Column(sql.String(64), nullable=False) name = sql.Column(sql.String(255), nullable=False) passwords = orm.relationship('Password', single_parent=True, cascade='all,delete-orphan', lazy='subquery', backref='local_user', order_by='Password.created_at') failed_auth_count = sql.Column(sql.Integer, nullable=True) failed_auth_at = sql.Column(sql.DateTime, nullable=True) __table_args__ = (sql.UniqueConstraint('domain_id', 'name'), {})
[docs]class Password(sql.ModelBase, sql.DictBase): __tablename__ = 'password' attributes = ['id', 'local_user_id', 'password', 'created_at', 'expires_at'] id = sql.Column(sql.Integer, primary_key=True) local_user_id = sql.Column(sql.Integer, sql.ForeignKey('local_user.id', ondelete='CASCADE')) password = sql.Column(sql.String(128), nullable=True) # created_at default set here to safe guard in case it gets missed created_at = sql.Column(sql.DateTime, nullable=False, default=datetime.datetime.utcnow) expires_at = sql.Column(sql.DateTime, nullable=True) self_service = sql.Column(sql.Boolean, default=False, nullable=False, server_default='0')
[docs]class FederatedUser(sql.ModelBase, sql.ModelDictMixin): __tablename__ = 'federated_user' attributes = ['id', 'user_id', 'idp_id', 'protocol_id', 'unique_id', 'display_name'] id = sql.Column(sql.Integer, primary_key=True) user_id = sql.Column(sql.String(64), sql.ForeignKey('user.id', ondelete='CASCADE')) idp_id = sql.Column(sql.String(64), sql.ForeignKey('identity_provider.id', ondelete='CASCADE')) protocol_id = sql.Column(sql.String(64), nullable=False) unique_id = sql.Column(sql.String(255), nullable=False) display_name = sql.Column(sql.String(255), nullable=True) __table_args__ = ( sql.UniqueConstraint('idp_id', 'protocol_id', 'unique_id'), sqlalchemy.ForeignKeyConstraint(['protocol_id', 'idp_id'], ['federation_protocol.id', 'federation_protocol.idp_id']) )
[docs]class NonLocalUser(sql.ModelBase, sql.ModelDictMixin): """SQL data model for nonlocal users (LDAP and custom).""" __tablename__ = 'nonlocal_user' attributes = ['domain_id', 'name', 'user_id'] domain_id = sql.Column(sql.String(64), primary_key=True) name = sql.Column(sql.String(255), primary_key=True) user_id = sql.Column(sql.String(64), sql.ForeignKey('user.id', ondelete='CASCADE'))
[docs]class Group(sql.ModelBase, sql.DictBase): __tablename__ = 'group' attributes = ['id', 'name', 'domain_id', 'description'] id = sql.Column(sql.String(64), primary_key=True) name = sql.Column(sql.String(64), nullable=False) domain_id = sql.Column(sql.String(64), nullable=False) description = sql.Column(sql.Text()) extra = sql.Column(sql.JsonBlob()) # Unique constraint across two columns to create the separation # rather than just only 'name' being unique __table_args__ = (sql.UniqueConstraint('domain_id', 'name'),)
[docs]class UserGroupMembership(sql.ModelBase, sql.DictBase): """Group membership join table.""" __tablename__ = 'user_group_membership' user_id = sql.Column(sql.String(64), sql.ForeignKey('user.id'), primary_key=True) group_id = sql.Column(sql.String(64), sql.ForeignKey('group.id'), primary_key=True)

Project Source