登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›OpenStack Keystone详解

OpenStack Keystone详解

诸婉丽 2025-9-12 10:59:49
Keystone 是 OpenStack 的核心组件之一，作为身份认证服务（Identity Service），它为整个 OpenStack 生态系统提供统一的身份验证、授权和服务目录管理。
三大核心功能：

  • 身份认证（Authentication）：验证用户身份（如用户名 / 密码、令牌等）
  • 授权（Authorization）：确定用户是否有权限执行特定操作
  • 服务目录（Service Catalog）：管理 OpenStack 各服务的访问端点（Endpoint）
1、核心概念

1.1 User（用户）

代表一个使用 OpenStack 服务的个体、系统或服务。它拥有登录凭证（密码、API Key 等）。
1.2 Project / Tenant（项目 / 租户）

Project 用于将 OpenStack 的资源（计算、存储和网络）进行分组和隔离。
根据 OpenStack 服务的对象不同，Project 可以是一个客户（公有云，也叫租户）、部门或者项目组（私有云）。
注意：

  • 资源的所有权是属于 Project 的，而不是 User。
  • 每个 User（包括 admin）必须挂在 Project 里才能访问该 Project 的资源。 一个User可以属于多个 Project。
  • admin 相当于 root 用户，具有最高权限
资源隔离的单元，用户必须属于某个项目才能操作资源(Tenant 是旧称，现在多称为 Project)
1.3 Role（角色）

权限的集合。它定义了一组用户可以执行的操作权限（如 member, admin, network-admin）。角色本身没有权限，其权限由各个 OpenStack 服务（Nova, Neutron等）在策略文件（policy.json）中定义。
示例：
  1. {   //用户需具有 admin角色 //
  2.     "context_is_admin": "role:admin",
  3.     //有管理员权限或请求用户的project_id和资源的project_id匹配 //
  4.     "admin_or_owner": "is_admin:True or project_id:%(project_id)s",
  5.     //检查用户上下文是否具有管理员权限。//
  6.     "admin_api": "is_admin:True",
  7.     //用户需具有 cloud_admin 角色 //
  8.     "cloud_admin": "role:cloud_admin",
  10.     // 引用规则admin_or_owner//
  11.     "os_compute_api:servers:start": "rule:admin_or_owner",
  13.     // 允许任何用户查看服务器元数据（metadata）
  14.     "os_compute_api:servers:show_server_metadata": ""
  15. }
复制代码
1.4 Token（令牌）

一个加密字符串，是访问资源的凭证。用户认证后，Keystone 会颁发一个 Token。在后续调用其他服务 API 时，必须出示此 Token 以验证身份和权限。Token 有有效期。
1.5 Service（服务）

代表一个 OpenStack 服务，如 nova, neutron, cinder 等。在 Keystone 中注册
1.6 Endpoint（端点）

服务的访问地址（URL），分为 public（公网）、internal（内网）、admin（管理）三种类型一个服务对外暴露的可访问地址（URL）。
1.7 Catalog（服务目录）

所有 Service 和其 Endpoint 的列表。它在用户认证成功后随 Token 一起返回，告诉用户哪里可以访问所需的服务。
1.8 Domain（域）

项目、用户和组的集合。用于实现更高级别的隔离和管理隔离。一个域可以被认为是一个更大的组织或部门，其下包含多个项目和用户。默认有一个 Default 域。
1.9 Group（组）

用户的集合。可以给一个组分配角色，那么组内的所有用户都会继承这个角色在项目中的权限，简化了用户权限管理。
2、工作原理

sequenceDiagram    actor User    participant H as Horizon/CLI    participant K as Keystone    participant S as Other Service
(e.g., Nova)    User->>H: 1. 提供用户名/密码
和项目名称    H->>K: 2. 认证请求
(用户名、密码、项目)    K->>K: 3. 验证凭证
计算权限
生成Token
组装服务目录    K-->>H: 4. 返回认证响应
(Token & Service Catalog)    H->>S: 5. 请求创建虚拟机
(携带Token)    S->>K: 6. 验证Token有效性    K-->>S: 7. 返回Token详情
(用户、项目、角色)    S->>S: 8. 根据策略(policy.json)
校验权限    S-->>H: 9. 执行操作并返回结果    H-->>User: 10. 显示操作结果

  • 用户认证 (Authentication): 用户（通过 Horizon 或 CLI）向 Keystone 提供用户名、密码和要访问的项目名称。
  • 验证与令牌颁发: Keystone 验证用户身份及其在该项目中的成员资格。验证通过后，Keystone 会：

    • 生成一个 Token。
    • 根据用户的角色计算其权限。
    • 组装该用户可以访问的服务目录 (Catalog)。

  • 服务访问: 用户使用收到的 Token 去调用其他 OpenStack 服务（例如 Nova 创建虚拟机）。该请求中必须包含 X-Auth-Token 头信息。
  • 令牌验证 (Validation): Nova 接收到请求后，无法自己识别 Token，它会向 Keystone 发出请求，询问“这个 Token 是否有效？它对应哪个用户和项目？”
  • 授权决策 (Authorization): Keystone 验证 Token 并返回Token的详情（用户、项目、角色等）。Nova 然后根据自己的policy.json文件，判断该用户在该项目下拥有的角色是否有权执行创建虚拟机操作。
  • 执行操作: 如果授权通过，Nova 执行操作；否则返回权限错误。
3、Keystone源码解析

Keystone 的源码主要位于 /keystone/ 目录下。
3.1 主要目录结构


  • keystone/common/: 通用工具（如配置、策略执行、上下文）
  • keystone/identity/: 身份后端驱动（SQL, LDAP）
  • keystone/assignment/: 资源（项目/域）和角色分配后端驱动
  • keystone/auth/: 认证插件（密码、令牌等）
  • keystone/token/: 令牌提供者（Fernet, UUID, PKI, PKIZ）
  • keystone/catalog/: 服务目录后端驱动
  • keystone/api/: 主要的 WSGI 路由和控制器（RESTful API 入口）
  • keystone/server/: 启动和部署相关的 Flask 应用配置
3.2 核心流程源码分析

3.2.1 密码认证与令牌颁发流程


  • 请求入口：WSGI 路由与控制器
    文件keystone/auth/routers.py
  1. class AuthRouter(wsgi.ComposableRouter):
  2.     def add_routes(self, mapper):
  3.         # 注册认证路由
  4.         mapper.connect('/auth/tokens',
  5.                        controller=self._controllers['auth'],
  6.                        action='authenticate_for_token',
  7.                        conditions={'method': ['POST']})
复制代码
当收到 POST /v3/auth/tokens请求时，路由将请求分发给 AuthController.authenticate_for_token方法。
2. 认证控制器处理
文件路径: keystone/auth/controllers.py
  1. class AuthController(controller.V3Controller):
  2.     def authenticate_for_token(self, context, auth_payload):
  3.         # 1. 验证请求格式
  4.         self._validate_authentication_request(context, auth_payload)
  5.         
  6.         # 2. 提取认证方法
  7.         auth_methods = auth_payload['auth']['identity']['methods']
  8.         auth_data = auth_payload['auth']['identity'].get('password', {})
  9.         
  10.         # 3. 选择认证插件
  11.         auth_plugin = self._get_auth_plugin(auth_methods[0])
  12.         
  13.         # 4. 执行认证
  14.         auth_context = auth_plugin.authenticate(context, **auth_data)
  15.         
  16.         # 5. 颁发令牌
  17.         token_data = self.token_provider_api.issue_token(
  18.             user_id=auth_context['user_id'],
  19.             method_names=auth_methods,
  20.             project_id=auth_context.get('project_id'),
  21.             domain_id=auth_context.get('domain_id')
  22.         )
  23.         
  24.         # 6. 构建响应
  25.         response = wsgi.render_response(
  26.             body=token_data,
  27.             status=(201, 'Created')
  28.         )
  29.         response.headers['X-Subject-Token'] = token_data['token']['id']
  30.         return response
复制代码

  • 密码认证插件实现
    文件路径: keystone/auth/plugins/password.py
  1. class Password(controller.AuthMethodHandler):
  2.     def authenticate(self, context, user_id=None, password=None,
  3.                      user_domain_id=None, username=None):
  4.         # 1. 获取用户域ID（如果未提供）
  5.         if not user_domain_id:
  6.             user_domain_id = CONF.identity.default_domain_id
  7.         
  8.         # 2. 获取用户对象
  9.         try:
  10.             if user_id:
  11.                 user_ref = self.identity_api.get_user(context, user_id)
  12.             else:
  13.                 user_ref = self.identity_api.get_user_by_name(
  14.                     context, username, user_domain_id)
  15.         except exception.UserNotFound:
  16.             raise exception.Unauthorized('Invalid user')
  17.         
  18.         # 3. 验证用户状态
  19.         if not user_ref.get('enabled', True):
  20.             raise exception.Unauthorized('User disabled')
  21.         
  22.         # 4. 验证密码
  23.         try:
  24.             self.identity_api.authenticate(
  25.                 context,
  26.                 user_id=user_ref['id'],
  27.                 password=password
  28.             )
  29.         except AssertionError:
  30.             raise exception.Unauthorized('Invalid password')
  31.         
  32.         # 5. 返回认证上下文
  33.         return {
  34.             'user_id': user_ref['id'],
  35.             'user_domain_id': user_domain_id,
  36.             'project_id': self._get_project_id(context, auth_context),
  37.             'domain_id': self._get_domain_id(context, auth_context)
  38.         }
复制代码

  • 身份驱动验证密码
    文件路径: keystone/identity/drivers/sql.py
  1. class Identity(identity.Driver):
  2.     def authenticate(self, context, user_id, password):
  3.         # 1. 获取用户凭证
  4.         try:
  5.             cred_ref = self.get_credential(context, user_id)
  6.         except exception.CredentialNotFound:
  7.             raise AssertionError("Invalid credentials")
  8.         
  9.         # 2. 验证密码哈希
  10.         if not self._check_password(password, cred_ref['blob']):
  11.             raise AssertionError("Invalid credentials")
  12.         
  13.         # 3. 验证凭证状态
  14.         if not cred_ref.get('enabled', True):
  15.             raise AssertionError("Credential disabled")
复制代码

  • 令牌提供者颁发令牌
    文件路径: keystone/token/provider.py
  1. class TokenProviderAPI(object):
  2.     def issue_token(self, user_id, method_names, **kwargs):
  3.         # 1. 验证用户和项目/域
  4.         self._assert_valid_user(user_id)
  5.         self._assert_valid_project_or_domain(kwargs)
  6.         
  7.         # 2. 生成令牌ID
  8.         token_id = self.token_formatter.create_token(
  9.             user_id=user_id,
  10.             **kwargs
  11.         )
  12.         
  13.         # 3. 构建令牌数据
  14.         token_data = {
  15.             'token': {
  16.                 'methods': method_names,
  17.                 'expires_at': self._get_expiration_time(),
  18.                 'user': self._get_user_data(user_id),
  19.                 'project': self._get_project_data(kwargs.get('project_id')),
  20.                 'domain': self._get_domain_data(kwargs.get('domain_id')),
  21.                 'catalog': self._get_service_catalog(user_id, **kwargs)
  22.             }
  23.         }
  24.         
  25.         # 4. 持久化令牌
  26.         self.token_provider.persist_token(token_id, token_data)
  27.         
  28.         return token_data
复制代码

  • Fernet 令牌驱动实现
    文件路径: keystone/token/providers/fernet.py
  1. class TokenFormatter(object):
  2.     def create_token(self, user_id, project_id=None, domain_id=None):
  3.         # 1. 准备令牌载荷
  4.         payload = {
  5.             'exp': int(time.time()) + CONF.token.expiration,
  6.             'iat': int(time.time()),
  7.             'user': {'id': user_id},
  8.             'aud': self._get_audience(),
  9.             'iss': CONF.token.issuer
  10.         }
  11.         
  12.         if project_id:
  13.             payload['project'] = {'id': project_id}
  14.         elif domain_id:
  15.             payload['domain'] = {'id': domain_id}
  16.         
  17.         # 2. 序列化载荷
  18.         serialized_payload = msgpack.packb(payload)
  19.         
  20.         # 3. 使用Fernet加密
  21.         fernet = Fernet(CONF.fernet_tokens.key_repository)
  22.         token = fernet.encrypt(serialized_payload)
  23.         
  24.         return token.decode('utf-8')
复制代码
3.2.2 令牌验证流程

其他服务调用 GET /v3/auth/tokens

  • 请求入口：路由与控制器
    文件路径: keystone/auth/routers.py
  1. class AuthRouter(wsgi.ComposableRouter):
  2.     def add_routes(self, mapper):
  3.         # 注册令牌验证路由
  4.         mapper.connect('/auth/tokens',
  5.                        controller=self._controllers['auth'],
  6.                        action='validate_token',
  7.                        conditions={'method': ['GET']})
复制代码

  • 令牌验证控制器
    文件路径: keystone/auth/controllers.py
  1. class AuthController(controller.V3Controller):
  2.     def validate_token(self, context):
  3.         # 1. 获取待验证的令牌ID
  4.         token_id = self._get_subject_token_id(context)
  5.         
  6.         # 2. 验证请求者权限
  7.         self.assert_authenticated()
  8.         
  9.         # 3. 调用令牌提供者API进行验证
  10.         token_data = self.token_provider_api.validate_token(token_id)
  11.         
  12.         # 4. 构建响应
  13.         return wsgi.render_response(body=token_data)
复制代码

  • 令牌提供者 API 实现
    文件路径: keystone/token/provider.py
  1. class TokenProviderAPI(object):
  2.     def validate_token(self, token_id):
  3.         # 1. 检查令牌格式
  4.         if not self.token_formatter.validate_token(token_id):
  5.             raise exception.ValidationError(_('Invalid token format'))
  6.         
  7.         # 2. 从缓存或持久化存储获取令牌数据
  8.         token_data = self._get_token_data(token_id)
  9.         
  10.         # 3. 验证令牌状态
  11.         self._validate_token_state(token_data)
  12.         
  13.         # 4. 验证令牌有效期
  14.         self._validate_token_expiration(token_data)
  15.         
  16.         # 5. 验证用户状态
  17.         self._assert_valid_user(token_data['token']['user']['id'])
  18.         
  19.         # 6. 验证项目/域状态
  20.         if 'project' in token_data['token']:
  21.             self._assert_valid_project(token_data['token']['project']['id'])
  22.         elif 'domain' in token_data['token']:
  23.             self._assert_valid_domain(token_data['token']['domain']['id'])
  24.         
  25.         # 7. 构建响应数据
  26.         return self._format_token_response(token_data)
复制代码

  • 令牌数据获取逻辑
    文件路径: keystone/token/provider.py
  1. class TokenProviderAPI(object):
  2.     def _get_token_data(self, token_id):
  3.         # 1. 检查缓存
  4.         token_data = self.token_cache.get_token(token_id)
  5.         if token_data:
  6.             return token_data
  7.         
  8.         # 2. 从持久化存储获取
  9.         token_data = self.token_driver.get_token_data(token_id)
  10.         if not token_data:
  11.             raise exception.TokenNotFound(_('Token not found'))
  12.         
  13.         # 3. 验证令牌签名
  14.         if not self.token_formatter.validate_token_signature(token_id):
  15.             raise exception.Unauthorized(_('Invalid token signature'))
  16.         
  17.         # 4. 缓存结果
  18.         self.token_cache.set_token(token_id, token_data)
  19.         
  20.         return token_data
复制代码

  • Fernet 令牌验证实现
    文件路径: keystone/token/providers/fernet.py
  1. class TokenFormatter(object):
  2.     def validate_token(self, token_id):
  3.         # 1. 基本格式检查
  4.         if not token_id or len(token_id) < CONF.fernet_tokens.min_token_size:
  5.             return False
  6.         
  7.         # 2. Fernet 令牌格式检查
  8.         return token_id.count('-') == 3 and len(token_id.split('-')[0]) == 8
  9.    
  10.     def validate_token_signature(self, token_id):
  11.         # 1. 尝试使用所有可用密钥解密
  12.         for key in self._get_valid_keys():
  13.             try:
  14.                 fernet = Fernet(key)
  15.                 payload = fernet.decrypt(token_id.encode('utf-8'))
  16.                 msgpack.unpackb(payload)  # 验证载荷格式
  17.                 return True
  18.             except (cryptography.fernet.InvalidToken, msgpack.UnpackException):
  19.                 continue
  20.         
  21.         return False
  22.    
  23.     def _get_valid_keys(self):
  24.         """获取所有有效的加密密钥"""
  25.         keys = []
  26.         for key_repo in CONF.fernet_tokens.key_repository:
  27.             for key_file in os.listdir(key_repo):
  28.                 if key_file.endswith('.key'):
  29.                     with open(os.path.join(key_repo, key_file), 'rb') as f:
  30.                         keys.append(f.read().strip())
  31.         return keys
复制代码

  • 令牌缓存实现
    文件路径: keystone/token/providers/common.py
  1. class TokenCache(object):
  2.     def __init__(self):
  3.         self._cache = {}
  4.         self._lock = threading.Lock()
  5.    
  6.     def get_token(self, token_id):
  7.         with self._lock:
  8.             # 1. 检查内存缓存
  9.             if token_id in self._cache:
  10.                 entry = self._cache[token_id]
  11.                 if time.time() < entry['exp']:
  12.                     return entry['data']
  13.                 del self._cache[token_id]
  14.             
  15.             # 2. 检查外部缓存（如 Memcached）
  16.             if CONF.token.caching and CONF.cache.enabled:
  17.                 cache_key = f"token_{token_id}"
  18.                 cached = cache.get(cache_key)
  19.                 if cached and time.time() < cached['exp']:
  20.                     self._cache[token_id] = cached  # 填充内存缓存
  21.                     return cached['data']
  22.         
  23.         return None
  24.    
  25.     def set_token(self, token_id, token_data):
  26.         with self._lock:
  27.             # 1. 设置内存缓存
  28.             entry = {
  29.                 'data': token_data,
  30.                 'exp': time.time() + CONF.token.cache_time
  31.             }
  32.             self._cache[token_id] = entry
  33.             
  34.             # 2. 设置外部缓存
  35.             if CONF.token.caching and CONF.cache.enabled:
  36.                 cache_key = f"token_{token_id}"
  37.                 cache.set(cache_key, entry, CONF.token.cache_time)
复制代码
3.2.3 服务注册


  • 服务注册 API 入口
    文件路径: keystone/service/routers.py
  1. class ServiceRouter(wsgi.ComposableRouter):
  2.     def add_routes(self, mapper):
  3.         # 服务注册路由
  4.         mapper.connect('/services',
  5.                        controller=self._controllers['service'],
  6.                        action='create_service',
  7.                        conditions={'method': ['POST']})
  8.         
  9.         # 端点注册路由
  10.         mapper.connect('/endpoints',
  11.                        controller=self._controllers['endpoint'],
  12.                        action='create_endpoint',
  13.                        conditions={'method': ['POST']})
复制代码

  • 服务控制器实现
    文件路径: keystone/service/controllers.py
  1. class ServiceController(controller.V3Controller):
  2.     @controller.protected()
  3.     def create_service(self, context, service):
  4.         """创建服务实体"""
  5.         # 1. 验证输入数据
  6.         validation.lazy_validate(schema.service_create, service)
  7.         service_ref = service['service']
  8.         
  9.         # 2. 生成服务ID
  10.         service_id = service_ref.get('id') or uuid.uuid4().hex
  11.         
  12.         # 3. 提取服务属性
  13.         service_type = service_ref['type']
  14.         service_name = service_ref.get('name', service_type)
  15.         description = service_ref.get('description', '')
  16.         enabled = service_ref.get('enabled', True)
  17.         
  18.         # 4. 调用服务API创建服务
  19.         service_dict = self.catalog_api.create_service(
  20.             service_id,
  21.             {'name': service_name, 'type': service_type,
  22.              'description': description, 'enabled': enabled}
  23.         )
  24.         
  25.         # 5. 返回创建的服务信息
  26.         return {'service': service_dict}
复制代码

  • 端点控制器实现
    文件路径: keystone/service/controllers.py
  1. class EndpointController(controller.V3Controller):
  2.     @controller.protected()
  3.     def create_endpoint(self, context, endpoint):
  4.         """创建服务端点"""
  5.         # 1. 验证输入数据
  6.         validation.lazy_validate(schema.endpoint_create, endpoint)
  7.         endpoint_ref = endpoint['endpoint']
  8.         
  9.         # 2. 生成端点ID
  10.         endpoint_id = endpoint_ref.get('id') or uuid.uuid4().hex
  11.         
  12.         # 3. 提取端点属性
  13.         service_id = endpoint_ref['service_id']
  14.         interface = endpoint_ref['interface']  # public, admin, internal
  15.         url = endpoint_ref['url']
  16.         region_id = endpoint_ref.get('region_id')
  17.         enabled = endpoint_ref.get('enabled', True)
  18.         
  19.         # 4. 验证服务是否存在
  20.         self.catalog_api.get_service(service_id)
  21.         
  22.         # 5. 调用端点API创建端点
  23.         endpoint_dict = self.catalog_api.create_endpoint(
  24.             endpoint_id,
  25.             {'service_id': service_id, 'interface': interface,
  26.              'url': url, 'region_id': region_id, 'enabled': enabled}
  27.         )
  28.         
  29.         # 6. 返回创建的端点信息
  30.         return {'endpoint': endpoint_dict}
复制代码

  • 目录服务 API 实现
    文件路径: keystone/catalog/core.py
  1. class CatalogManager(object):
  2.     def __init__(self):
  3.         # 加载后端驱动
  4.         self.driver = self._get_driver()
  5.         
  6.     def _get_driver(self):
  7.         # 根据配置选择驱动
  8.         if CONF.catalog.driver == 'sql':
  9.             from keystone.catalog.backends import sql
  10.             return sql.Catalog()
  11.         elif CONF.catalog.driver == 'templated':
  12.             from keystone.catalog.backends import templated
  13.             return templated.Catalog()
  14.         else:
  15.             raise exception.ConfigError('Invalid catalog driver')
  16.    
  17.     def create_service(self, service_id, service):
  18.         """创建服务实体"""
  19.         # 验证服务类型
  20.         self._validate_service_type(service['type'])
  21.         
  22.         # 调用驱动创建服务
  23.         return self.driver.create_service(service_id, service)
  24.    
  25.     def create_endpoint(self, endpoint_id, endpoint):
  26.         """创建服务端点"""
  27.         # 验证接口类型
  28.         self._validate_interface(endpoint['interface'])
  29.         
  30.         # 调用驱动创建端点
  31.         return self.driver.create_endpoint(endpoint_id, endpoint)
复制代码

  • SQL 后端驱动实现
    文件路径: keystone/catalog/backends/sql.py
  1. class Catalog(DriverBase):
  2.     def create_service(self, service_id, service):
  3.         """在SQL数据库中创建服务"""
  4.         with sql.session_for_write() as session:
  5.             # 检查服务类型是否唯一
  6.             if session.query(models.Service).filter_by(
  7.                 type=service['type']).first():
  8.                 raise exception.Conflict('Service type already exists')
  9.             
  10.             # 创建服务对象
  11.             service_ref = models.Service()
  12.             service_ref.update({
  13.                 'id': service_id,
  14.                 'name': service['name'],
  15.                 'type': service['type'],
  16.                 'description': service.get('description'),
  17.                 'enabled': service.get('enabled', True)
  18.             })
  19.             
  20.             # 添加到会话并提交
  21.             session.add(service_ref)
  22.             session.flush()
  23.             
  24.             return service_ref.to_dict()
  25.    
  26.     def create_endpoint(self, endpoint_id, endpoint):
  27.         """在SQL数据库中创建端点"""
  28.         with sql.session_for_write() as session:
  29.             # 检查服务是否存在
  30.             service_ref = session.query(models.Service).get(
  31.                 endpoint['service_id'])
  32.             if not service_ref:
  33.                 raise exception.ServiceNotFound(
  34.                     service_id=endpoint['service_id'])
  35.             
  36.             # 检查接口类型是否重复
  37.             if session.query(models.Endpoint).filter_by(
  38.                 service_id=endpoint['service_id'],
  39.                 interface=endpoint['interface']).first():
  40.                 raise exception.Conflict(
  41.                     'Endpoint interface already exists for this service')
  42.             
  43.             # 创建端点对象
  44.             endpoint_ref = models.Endpoint()
  45.             endpoint_ref.update({
  46.                 'id': endpoint_id,
  47.                 'service_id': endpoint['service_id'],
  48.                 'interface': endpoint['interface'],
  49.                 'url': endpoint['url'],
  50.                 'region_id': endpoint.get('region_id'),
  51.                 'enabled': endpoint.get('enabled', True)
  52.             })
  53.             
  54.             # 添加到会话并提交
  55.             session.add(endpoint_ref)
  56.             session.flush()
  57.             
  58.             return endpoint_ref.to_dict()
复制代码

  • 模板化后端驱动实现
    文件路径: keystone/catalog/backends/templated.py
  1. class Catalog(DriverBase):
  2.     def __init__(self):
  3.         # 加载配置文件
  4.         self.conf = self._load_config()
  5.         
  6.     def _load_config(self):
  7.         """从配置文件加载服务目录模板"""
  8.         with open(CONF.catalog.template_file, 'r') as f:
  9.             return json.load(f)
  10.    
  11.     def create_service(self, service_id, service):
  12.         """模板驱动不支持动态创建服务"""
  13.         raise exception.NotImplemented(
  14.             'Dynamic service creation not supported in templated driver')
  15.    
  16.     def create_endpoint(self, endpoint_id, endpoint):
  17.         """模板驱动不支持动态创建端点"""
  18.         raise exception.NotImplemented(
  19.             'Dynamic endpoint creation not supported in templated driver')
  20.    
  21.     def get_catalog(self, user_id, project_id):
  22.         """从模板生成服务目录"""
  23.         # 替换模板中的变量
  24.         catalog = copy.deepcopy(self.conf['catalog'])
  25.         for service in catalog:
  26.             for endpoint in service['endpoints']:
  27.                 endpoint['url'] = endpoint['url'].replace(
  28.                     '$HOST', CONF.default_public_host)
  29.         return catalog
复制代码
4、常用命令

4.1 身份（Identity）管理

1. 身份（Identity）管理

  • 创建用户:
    1. openstack user create --password-prompt --email user@example.com --domain Default myuser
    复制代码

    • --password-prompt 会交互式让你输入密码。
    • --domain 指定用户所在的域。

  • 列出用户:
    1. openstack user list
    复制代码
  • 创建用户组 (用于LDAP集成或批量分配角色):
    1. openstack group create --description "Developers Group" developers
    复制代码
  • 将用户加入用户组:
    1. openstack group add user developers myuser
    复制代码
4.2 资源（Resource）与分配（Assignment）管理


  • 创建项目（Project）:
    1. openstack project create --description "Demo Project" --domain Default demoproject
    复制代码
  • 创建角色（Role）:
    1. openstack role create myrole
    复制代码

    • 通常内置 admin, _member_, reader 等角色。

  • 给用户在项目上授予角色 (核心操作):
    1. openstack role add --project demoproject --user myuser myrole
    复制代码

    • 这条命令意味着 myuser 在 demoproject 中拥有 myrole 的权限。

  • 检查角色分配:
    1. openstack role assignment list --user myuser --project demoproject --names
    复制代码
4.3 服务目录（Catalog）管理


  • 注册服务 (如注册Glance镜像服务):
    1. openstack service create --name glance --description "OpenStack Image" image
    复制代码
  • 创建服务的端点（Endpoint）:
    1. openstack endpoint create --region RegionOne image public http://glance-host:9292
    2. openstack endpoint create --region RegionOne image internal http://glance-host:9292
    3. openstack endpoint create --region RegionOne image admin http://glance-host:9292
    复制代码
  • 列出所有服务和端点:
    1. openstack service list
    2. openstack endpoint list
    复制代码
4.4 令牌（Token）操作


  • 作为特定用户获取令牌（调试神器）:
    1. openstack --os-username myuser --os-password mypassword --os-project-name demoproject token issue
    复制代码

    • 这会返回一个详细的令牌信息，包括过期时间、授权范围等，用于调试权限问题。

4.5 域（Domain）管理


  • 创建域:
    1. openstack domain create --description "My Company" mycompany
    复制代码
  • 在特定域中创建用户:
    1. openstack user create --email john@mycompany.com --domain mycompany john
    复制代码

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
OpenStackKeystone详解

使用道具 举报

相关推荐
许娴广
2025-10-10 00:38:40

举报

喜欢鼓捣这些软件，现在用得少，谢谢分享！
缑莺韵
2025-10-14 01:52:55

举报

谢谢楼主提供！
命煦砌
12 小时前

举报

分享、互助 让互联网精神温暖你我
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
诸婉丽
12 小时前

0

粉丝关注

25

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫