openstack/codegenerator
Code Issues Proposed changes

Add barbican secret schemas

Browse Source 
Sadly there is not normal way to extract usable schemas from barbican. Hardcode
them.

Change-Id: Iddeaad08d82c64c7763079704f65126307271f47
This commit is contained in:
Artem Goncharov 2025-02-11 13:12:51 +01:00
parent 8995fa42b2
commit 3aa89ec1da
5 changed files with 510 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
codegenerator/cli.py
View File

@ -173,7 +173,7 @@ def main():
)
parser.add_argument(
"--alternative-module-name",
help=("Optional new module name " "(rename get into list)"),
help=("Optional new module name (rename get into list)"),
)
parser.add_argument(
"--openapi-yaml-spec", help=("Path to the OpenAPI spec file (yaml)")

48
codegenerator/openapi/barbican.py
View File

@ -20,6 +20,7 @@ import fixtures
from codegenerator.common.schema import SpecSchema
from codegenerator.openapi.base import OpenStackServerSourceBase
from codegenerator.openapi.utils import merge_api_ref_doc
from codegenerator.openapi.barbican_schemas import secret
from ruamel.yaml.scalarstring import LiteralScalarString
@ -27,6 +28,8 @@ from ruamel.yaml.scalarstring import LiteralScalarString
class BarbicanGenerator(OpenStackServerSourceBase):
URL_TAG_MAP = {}
RESOURCE_MODULES = [secret]
def __init__(self):
pass
@ -259,7 +262,7 @@ class BarbicanGenerator(OpenStackServerSourceBase):
None,
"/v1/secrets/{secret_id}/metadata",
controller=secretmeta.SecretMetadataController.on_get,
action="get",
action="list",
conditions={"method": ["GET"]},
)
mapper.connect(
@ -281,14 +284,14 @@ class BarbicanGenerator(OpenStackServerSourceBase):
None,
"/v1/secrets/{secret_id}/metadata/{key}",
controller=secretmeta.SecretMetadatumController.on_get,
action="get",
action="show",
conditions={"method": ["GET"]},
)
mapper.connect(
None,
"/v1/secrets/{secret_id}/metadata/{key}",
controller=secretmeta.SecretMetadatumController.on_put,
action="create",
action="update",
conditions={"method": ["PUT"]},
)
mapper.connect(
@ -303,7 +306,7 @@ class BarbicanGenerator(OpenStackServerSourceBase):
None,
"/v1/secrets/{secret_id}/consumers",
controller=consumers.SecretConsumersController.on_get,
action="get",
action="list",
conditions={"method": ["GET"]},
)
mapper.connect(
@ -460,3 +463,40 @@ class BarbicanGenerator(OpenStackServerSourceBase):
lnk.symlink_to(impl_path.name)
return impl_path
def _post_process_operation_hook(
self, openapi_spec, operation_spec, path: str | None = None
):
"""Hook to allow service specific generator to modify details"""
for resource_mod in self.RESOURCE_MODULES:
hook = getattr(resource_mod, "_post_process_operation_hook", None)
if hook:
hook(openapi_spec, operation_spec, path=path)
def _get_schema_ref(
self,
openapi_spec,
name,
description=None,
schema_def=None,
action_name=None,
):
# Invoke modularized schema _get_schema_ref
for resource_mod in self.RESOURCE_MODULES:
hook = getattr(resource_mod, "_get_schema_ref", None)
if hook:
(ref, mime_type, matched) = hook(
openapi_spec, name, description, schema_def, action_name
)
if matched:
return (ref, mime_type)
# Default
(ref, mime_type) = super()._get_schema_ref(
openapi_spec,
name,
description,
schema_def=schema_def,
action_name=action_name,
)
return (ref, mime_type)

464
codegenerator/openapi/barbican_schemas/secret.py Normal file
View File

@ -0,0 +1,464 @@
# 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.
#
from typing import Any
from barbican.common import validators
from barbican.plugin.interface import secret_store
from codegenerator.common.schema import ParameterSchema
from codegenerator.common.schema import TypeSchema
SECRET_CONSUMER_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {
"service": {"type": "string", "description": "Consume service type"},
"resource_type": {
"type": "string",
"description": "Name of the resource type using the secret e.g. 'images' or 'lbaas/loadbalancers'",
},
"resource_id": {
"type": "string",
"description": "Unique identifier for the resource using this secret.",
},
},
"required": ["service", "resource_type", "resource_id"],
}
SECRET_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {
"name": {
"type": ["string", "null"],
"description": "The name of the secret set by the user.",
},
"algorithm": {
"type": ["string", "null"],
"description": "Metadata provided by a user or system for informational purposes.",
},
"mode": {
"type": ["string", "null"],
"description": "Metadata provided by a user or system for informational purposes.",
},
"bit_length": {
"type": ["integer", "null"],
"description": "Metadata provided by a user or system for informational purposes. ",
},
"expiration": {
"type": ["string", "null"],
"description": "This is a UTC timestamp in ISO 8601 format YYYY-MM-DDTHH:MM:SSZ. If set, the secret will not be available after this time.",
},
"secret_ref": {
"type": "string",
"description": "Reference of the new secret",
},
"secret_type": {
"type": ["string", "null"],
"enum": [
secret_store.SecretType.SYMMETRIC,
secret_store.SecretType.PASSPHRASE,
secret_store.SecretType.PRIVATE,
secret_store.SecretType.PUBLIC,
secret_store.SecretType.CERTIFICATE,
secret_store.SecretType.OPAQUE,
],
"description": "Used to indicate the type of secret being stored.",
},
"content-types": {
"type": "object",
"additionalProperties": {"type": "string"},
},
"status": {"type": "string", "description": "Secret status"},
"created": {"type": "string", "format": "date-time"},
"updated": {"type": ["string", "null"], "format": "date-time"},
"expiration": {"type": ["string", "null"], "format": "date-time"},
"consumers": {"type": "array", "items": SECRET_CONSUMER_SCHEMA},
},
"additionalProperties": False,
}
SECRET_LIST_RESPONSE: dict[str, Any] = {
"type": "object",
"properties": {
"next": {
"type": "string",
"description": "A HATEOAS URL to retrieve the next set of secrets based on the offset and limit parameters. This attribute is only available when the total number of secrets is greater than offset and limit parameter combined.",
},
"previous": {
"type": "string",
"description": "A HATEOAS URL to retrieve the previous set of secrets based on the offset and limit parameters. This attribute is only available when the request offset is greater than 0.",
},
"total": {
"type": "integer",
"description": "The total number of secrets available to the user.",
},
"secrets": {
"type": "array",
"description": "Contains a list of secrets. The attributes in the secret objects are the same as for an individual secret.",
"items": SECRET_SCHEMA,
},
},
"required": ["secrets", "total"],
"additionalProperties": False,
}
SECRET_CREATE_REQUEST: dict[str, Any] = validators.NewSecretValidator().schema
SECRET_CREATE_RESPONSE: dict[str, Any] = {
"type": "object",
"properties": {
"secret_ref": {
"type": "string",
"description": "Reference of the new secret",
}
},
"additionalProperties": False,
}
SECRETS_LIST_PARAMETERS: dict[str, dict] = {
"offset": {
"in": "query",
"name": "offset",
"description": "The starting index within the total list of the secrets that you would like to retrieve.",
"schema": {"type": "integer"},
},
"limit": {
"in": "query",
"name": "limit",
"description": "The maximum number of records to return (up to 100). The default limit is 10.",
"schema": {"type": "integer"},
},
"name": {
"in": "query",
"name": "name",
"description": "Selects all secrets with name similar to this value.",
"schema": {"type": "string"},
},
"alg": {
"in": "query",
"name": "alg",
"description": "Selects all secrets with algorithm similar to this value.",
"schema": {"type": "string"},
},
"mode": {
"in": "query",
"name": "mode",
"description": "Selects all secrets with mode similar to this value.",
"schema": {"type": "string"},
},
"bits": {
"in": "query",
"name": "bits",
"description": "Selects all secrets with bit_length equal to this value.",
"schema": {"type": "integer"},
},
"secret_type": {
"in": "query",
"name": "secret_type",
"description": "Selects all secrets with secret_type equal to this value.",
"schema": {
"type": "string",
"enum": [
secret_store.SecretType.SYMMETRIC,
secret_store.SecretType.PASSPHRASE,
secret_store.SecretType.PRIVATE,
secret_store.SecretType.PUBLIC,
secret_store.SecretType.CERTIFICATE,
secret_store.SecretType.OPAQUE,
],
},
},
"acl_only": {
"in": "query",
"name": "acl_only",
"description": "Selects all secrets with an ACL that contains the user. Project scope is ignored.",
"schema": {"type": "boolean"},
},
"created": {
"in": "query",
"name": "created",
"description": "Date filter to select all secrets with created matching the specified criteria.",
"schema": {"type": "string"},
},
"updated": {
"in": "query",
"name": "updated",
"description": "Date filter to select all secrets with updated matching the specified criteria.",
"schema": {"type": "string"},
},
"expiration": {
"in": "query",
"name": "expiration",
"description": "Date filter to select all secrets with expiration matching the specified criteria.",
"schema": {"type": "string"},
},
"sort": {
"in": "query",
"name": "sort",
"description": "Determines the sorted order of the returned list.",
"schema": {"type": "string"},
},
}
SECRET_ACL_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {
"users": {
"type": "array",
"items": {"type": "string"},
"description": "List of user ids. This needs to be a user id as returned by Keystone.",
},
"project_access": {
"type": "boolean",
"description": "Flag to mark a secret private so that the user who created the secret and users specified in above list can only access the secret. Pass false to mark the secret private.",
},
},
"additionalProperties": False,
}
SECRET_ACL_CONTAINER_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {"read": SECRET_ACL_SCHEMA},
"required": ["read"],
"additionalProperties": False,
}
SECRET_ACL_CREATE_RESPONSE: dict[str, Any] = {
"type": "object",
"properties": {"acl_ref": {"type": "string"}},
"required": ["acl_ref"],
"additionalProperties": False,
}
SECRET_METADATA_SCHEMA: dict[str, Any] = {
"type": "object",
"additionalProperties": {"type": "string"},
}
SECRET_METADATA_CONTAINER_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {"metadata": SECRET_METADATA_SCHEMA},
"required": ["metadata"],
"additionalProperties": False,
}
SECRET_METADATA_CREATE_RESPONSE: dict[str, Any] = {
"type": "object",
"properties": {"metadata_ref": {"type": "string"}},
"required": ["metadata_ref"],
"additionalProperties": False,
}
SECRET_METADATA_KEY_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {"key": {"type": "string"}, "value": {"type": "string"}},
"required": ["key", "value"],
"additionalProperties": False,
}
SECRET_CONSUMER_RESPONSE_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {
**SECRET_CONSUMER_SCHEMA["properties"],
"status": {"type": "string"},
"created": {"type": "string", "format": "date-time"},
"updated": {"type": ["string", "null"], "format": "date-time"},
},
"required": ["service", "resource_type", "resource_id"],
"additionalProperties": False,
}
SECRET_CONSUMERS_RESPONSE: dict[str, Any] = {
"type": "object",
"properties": {
"next": {
"type": "string",
"description": "A HATEOAS URL to retrieve the next set of secrets based on the offset and limit parameters. This attribute is only available when the total number of secrets is greater than offset and limit parameter combined.",
},
"previous": {
"type": "string",
"description": "A HATEOAS URL to retrieve the previous set of secrets based on the offset and limit parameters. This attribute is only available when the request offset is greater than 0.",
},
"total": {
"type": "integer",
"description": "The total number of secrets available to the user.",
},
"consumers": {"type": "array", "items": SECRET_CONSUMER_SCHEMA},
},
"required": ["total", "consumers"],
"additionalProperties": False,
}
SECRETS_CONSUMERS_LIST_PARAMETERS: dict[str, dict] = {
"offset": {
"in": "query",
"name": "offset",
"description": "The starting index within the total list of the secrets that you would like to retrieve.",
"schema": {"type": "integer"},
},
"limit": {
"in": "query",
"name": "limit",
"description": "The maximum number of records to return (up to 100). The default limit is 10.",
"schema": {"type": "integer"},
},
}
def _post_process_operation_hook(
openapi_spec, operation_spec, path: str | None = None
):
"""Hook to allow service specific generator to modify details"""
operationId = operation_spec.operationId
if operationId == "secrets:get":
for key, val in SECRETS_LIST_PARAMETERS.items():
openapi_spec.components.parameters.setdefault(
key, ParameterSchema(**val)
)
ref = f"#/components/parameters/{key}"
if ref not in [x.ref for x in operation_spec.parameters]:
operation_spec.parameters.append(ParameterSchema(ref=ref))
elif operationId == "secrets/secret_id:put":
operation_spec.parameters.append(
ParameterSchema(
name="content-encoding",
type_schema=TypeSchema(type="string"),
location="header",
)
)
operation_spec.parameters.append(
ParameterSchema(
name="content-type",
type_schema=TypeSchema(type="string"),
location="header",
)
)
operation_spec.requestBody = {
"content": {
"*/*": {"schema": {"type": "string", "format": "binary"}}
}
}
operation_spec.responses = {"204": {"description": "ok"}}
elif operationId == "secrets/secret_id/payload:get":
operation_spec.responses = {
"200": {
"description": "ok",
"content": {
"*/*": {"schema": {"type": "string", "format": "binary"}}
},
}
}
elif operationId == "secrets/secret_id/consumers:get":
for key, val in SECRETS_CONSUMERS_LIST_PARAMETERS.items():
openapi_spec.components.parameters.setdefault(
key, ParameterSchema(**val)
)
ref = f"#/components/parameters/{key}"
if ref not in [x.ref for x in operation_spec.parameters]:
operation_spec.parameters.append(ParameterSchema(ref=ref))
elif operationId == "secrets/secret_id/consumers:delete":
operation_spec.requestBody = {
"content": {"application/json": {"schema": SECRET_CONSUMER_SCHEMA}}
}
operation_spec.responses = {
"200": {
"description": "ok",
"content": {"application/json": {"schema": SECRET_SCHEMA}},
}
}
def _get_schema_ref(
openapi_spec, name, description=None, schema_def=None, action_name=None
) -> tuple[str | None, str | None, bool]:
mime_type: str = "application/json"
ref: str
# Secret
if name == "SecretsListResponse":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_LIST_RESPONSE)
)
ref = f"#/components/schemas/{name}"
elif name == "SecretGetResponse":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_SCHEMA)
)
ref = f"#/components/schemas/{name}"
elif name == "SecretsCreateRequest":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_CREATE_REQUEST)
)
ref = f"#/components/schemas/{name}"
elif name == "SecretsCreateResponse":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_CREATE_RESPONSE)
)
ref = f"#/components/schemas/{name}"
elif name in [
"SecretsAclGetResponse",
"SecretsAclCreateRequest",
"SecretsAclUpdateRequest",
]:
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_ACL_CONTAINER_SCHEMA)
)
ref = f"#/components/schemas/{name}"
elif name in ["SecretsAclCreateResponse", "SecretsAclUpdateResponse"]:
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_ACL_CREATE_RESPONSE)
)
ref = f"#/components/schemas/{name}"
elif name in [
"SecretsMetadataListResponse",
"SecretsMetadataCreateRequest",
]:
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_METADATA_CONTAINER_SCHEMA)
)
ref = f"#/components/schemas/{name}"
elif name in ["SecretsMetadataCreateResponse"]:
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_METADATA_CREATE_RESPONSE)
)
ref = f"#/components/schemas/{name}"
elif name in [
"SecretsMetadataShowResponse",
"SecretsMetadataReplaceRequest",
"SecretsMetadataReplaceResponse",
"SecretsMetadataUpdateRequest",
"SecretsMetadataUpdateResponse",
]:
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_METADATA_KEY_SCHEMA)
)
ref = f"#/components/schemas/{name}"
elif name == "SecretsConsumersListResponse":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_CONSUMERS_RESPONSE)
)
ref = f"#/components/schemas/{name}"
elif name in ["SecretsConsumersCreateRequest"]:
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_CONSUMER_SCHEMA)
)
ref = f"#/components/schemas/{name}"
elif name in ["SecretsConsumersCreateResponse"]:
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**SECRET_SCHEMA)
)
ref = f"#/components/schemas/{name}"
else:
return (None, None, False)
return (ref, mime_type, True)

1
codegenerator/openapi/base.py
View File

@ -1381,7 +1381,6 @@ class OpenStackServerSourceBase:
getattr(f, "_response_body_schema", {}),
)
response_body_schema = obj
print(f"the response schema is {response_body_schema}")
if "query_params_schema" in closure_locals or hasattr(
f, "_request_query_schema"
):

2
codegenerator/openapi/neutron.py
View File

@ -905,7 +905,7 @@ class NeutronGenerator(OpenStackServerSourceBase):
"AgentsL3_RouterUpdateResponse",
"RoutersL3_AgentShowResponse",
"RoutersL3_AgentUpdateRequest",
"RoutersL3_AgentUpdateResponse" "RoutersL3_AgentsCreateRequest",
"RoutersL3_AgentUpdateResponseRoutersL3_AgentsCreateRequest",
"RoutersL3_AgentsCreateResponse",
]:
return None