Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • osl/code/addons/org.etsi.osl.controllers.camara
1 result
Show changes
Expand all Hide whitespace changes
Inline Side-by-side
Showing
with 2206 additions and 0 deletions
QoDProvisioning/QoDProvisioningAPI/API/src/config.py 0 → 100644
View file @ 64d0ca5b
# -*- coding: utf-8 -*-
# @Authors:
# Eduardo Santos (eduardosantoshf@av.it.pt)
# Rafael Direito (rdireito@av.it.pt)
# @Organization:
# Instituto de Telecomunicações, Aveiro (ITAv)
# Aveiro, Portugal
# @Date:
# December 2024
import os
import sys
import logging
class Config():
broker_address = os.getenv('BROKER_ADDRESS')
broker_port = os.getenv('BROKER_PORT')
broker_username = os.getenv('BROKER_USERNAME')
broker_password = os.getenv('BROKER_PASSWORD')
service_uuid = os.getenv('SERVICE_UUID')
log_level = os.getenv('LOG_LEVEL', "INFO")
db_path = os.getenv("SQLITE_DB_PATH", "/data/sqlite.db")
# Broker topics
catalog_upd_service = "CATALOG.UPD.SERVICE"
event_service_attrchanged = "EVENT.SERVICE.ATTRCHANGED"
logger = None
@classmethod
def validate(cls):
missing_envs = []
for var in ['broker_address', 'broker_port', 'broker_username',
'broker_password', 'service_uuid']:
if getattr(cls, var) is None:
missing_envs.append(var.upper())
if missing_envs:
raise EnvironmentError(
f"Missing required environment variables: {', '.join(missing_envs)}"
)
print("All required environment variables are set.")
@classmethod
def setup_logging(cls):
if cls.logger is None:
log_level = getattr(logging, cls.log_level.upper())
# Create a logger
cls.logger = logging.getLogger()
cls.logger.setLevel(log_level)
# Create a stream handler that outputs to stdout
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(log_level)
# Create a formatter and add it to the handler
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
# Add the handler to the logger
cls.logger.addHandler(handler)
return cls.logger
\ No newline at end of file
QoDProvisioning/QoDProvisioningAPI/API/src/database/base_models.py 0 → 100644
View file @ 64d0ca5b
# -*- coding: utf-8 -*-
# @Authors:
# Eduardo Santos (eduardosantoshf@av.it.pt)
# Rafael Direito (rdireito@av.it.pt)
# @Organization:
# Instituto de Telecomunicações, Aveiro (ITAv)
# Aveiro, Portugal
# @Date:
# December 2024
from sqlalchemy import (
Column,
String,
Integer,
DateTime,
ForeignKey,
Enum as SAEnum,
)
from sqlalchemy.orm import relationship, validates
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.ext.declarative import declarative_base
from database.db import Base, engine
from schemas.status import Status
from schemas.status_info import StatusInfo
from enum import Enum as PyEnum
import uuid
from datetime import datetime
import re
class Device(Base):
__tablename__ = 'device'
__table_args__ = {"extend_existing": True}
id = Column(Integer, primary_key=True)
phone_number = Column(String, nullable=True)
network_access_identifier = Column(String, nullable=True)
ipv4_public_address = Column(String, nullable=True)
ipv4_private_address = Column(String, nullable=True)
ipv4_public_port = Column(Integer, nullable=True)
ipv6_address = Column(String, nullable=True)
provisioning = relationship('Provisioning', back_populates='device')
class Provisioning(Base):
__tablename__ = 'provisioning'
__table_args__ = {"extend_existing": True}
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
qos_profile = Column(String(256), nullable=False)
sink = Column(String, nullable=True)
device_id = Column(Integer, ForeignKey('device.id'), nullable=True)
sink_credential = Column(String, nullable=True, default=None)
started_at = Column(DateTime, default=datetime.utcnow)
status = Column(SAEnum(Status), nullable=False, default=Status.REQUESTED)
status_info = Column(SAEnum(StatusInfo), nullable=True)
device = relationship('Device', back_populates='provisioning')
QoDProvisioning/QoDProvisioningAPI/API/src/database/crud.py 0 → 100644
View file @ 64d0ca5b
# -*- coding: utf-8 -*-
# @Authors:
# Eduardo Santos (eduardosantoshf@av.it.pt)
# Rafael Direito (rdireito@av.it.pt)
# @Organization:
# Instituto de Telecomunicações, Aveiro (ITAv)
# Aveiro, Portugal
# @Date:
# December 2024
from sqlalchemy.orm import Session
from database.db import SessionLocal
from sqlalchemy.exc import SQLAlchemyError
from typing import List
from datetime import datetime
from fastapi import HTTPException
from schemas.create_provisioning import CreateProvisioning
from database.base_models import Provisioning, Device
from schemas.provisioning_info import ProvisioningInfo
from schemas.status import Status
from schemas.status_changed import StatusChanged
from schemas.status_info import StatusInfo
from schemas.retrieve_provisioning_by_device import RetrieveProvisioningByDevice
from config import Config
# Set up logging
logger = Config.setup_logging()
def retrieve_fields_to_check(device: Device) -> list:
"""
Retrieves the list of fields to check for a given device.
Args:
device: The device object that contains the fields to be checked.
Returns:
A list of tuples, each containing the field name and its
corresponding value.
"""
return [
("phone_number", device.phone_number),
(
"ipv4_public_address",
device.ipv4_address.public_address if device.ipv4_address else None
),
(
"ipv4_private_address",
device.ipv4_address.private_address if device.ipv4_address else None
),
(
"ipv4_public_port",
device.ipv4_address.public_port if device.ipv4_address else None
),
("ipv6_address", device.ipv6_address),
("network_access_identifier", device.network_access_identifier)
]
def find_existing_device(db: Session, fields_to_check: list) -> Device:
"""
Find an existing device based on the fields provided.
Args:
db: Database session.
fields_to_check: List of tuples with field names and values.
Returns:
The existing device if found, None otherwise.
"""
for field, value in fields_to_check:
if value: # Only search if the field has a value
existing_device = db.query(Device).filter_by(**{field: value})\
.first()
if existing_device:
logger.debug(f"Existing device found: {existing_device}")
return existing_device
return None
def validate_device_fields(create_provisioning):
"""
Validates the fields in the device object and assigns them to variables.
Args:
create_provisioning: The provisioning object containing the
device to validate.
Returns:
A dictionary containing the validated fields.
"""
device = create_provisioning.device
# Validate phone_number and network_access_identifier
phone_number = device.phone_number if device.phone_number else None
network_access_identifier = device.network_access_identifier \
if device.network_access_identifier else None
# Validate ipv4_address and its subfields
ipv4_address = device.ipv4_address
if ipv4_address:
ipv4_public_address = ipv4_address.public_address \
if ipv4_address.public_address else None
ipv4_private_address = ipv4_address.private_address \
if ipv4_address.private_address else None
ipv4_public_port = ipv4_address.public_port \
if ipv4_address.public_port else None
else:
ipv4_public_address = None
ipv4_private_address = None
ipv4_public_port = None
# Validate ipv6_address
ipv6_address = device.ipv6_address if device.ipv6_address else None
# Return all the validated fields in a dictionary
return {
'phone_number': phone_number,
'network_access_identifier': network_access_identifier,
'ipv4_public_address': ipv4_public_address,
'ipv4_private_address': ipv4_private_address,
'ipv4_public_port': ipv4_public_port,
'ipv6_address': ipv6_address
}
def create_provisioning(
db: Session, create_provisioning: CreateProvisioning
) -> Provisioning:
"""
Creates a new provisioning in the database.
Args:
db: Database session.
create_provisioning: The data needed to create the provisioning.
Returns:
The created Provisioning object.
"""
try:
logger.debug(f"Received provisioning data: {create_provisioning}\n")
device = create_provisioning.device
fields_to_check = retrieve_fields_to_check(device)
# Find an existing device if any field matches
existing_device = find_existing_device(db, fields_to_check)
# If device exists, check for field differences
if existing_device:
# Compare provided fields with the existing device fields
differences = [
(field, value, getattr(existing_device, field) != value)
for field, value in fields_to_check
]
# Check if there's at least one difference,
# and whether a new field is being added
new_field_added = False
for field, value, differs in differences:
if differs:
existing_value = getattr(existing_device, field)
# Field doesn't exist in the existing device
if existing_value is None:
logger.debug(
f"Adding new field {field} to existing device."
)
setattr(existing_device, field, value)
new_field_added = True
else:
# If any field differs, raise a conflict
logger.debug(
f"Device already exists, but fields differ: {field}"
)
raise HTTPException(
status_code=409,
detail="Device already exists, but fields differ."
)
# If no differences found, reuse the existing device
new_device = existing_device
else:
# Validate fields
validated_fields = validate_device_fields(create_provisioning)
# Create a new device instance using the validated fields
new_device = Device(
phone_number=validated_fields['phone_number'],
network_access_identifier=validated_fields[
'network_access_identifier'
],
ipv4_public_address=validated_fields['ipv4_public_address'],
ipv4_private_address=validated_fields['ipv4_private_address'],
ipv4_public_port=validated_fields['ipv4_public_port'],
ipv6_address=validated_fields['ipv6_address']
)
# Add the new device to the session
db.add(new_device)
db.commit()
db.refresh(new_device)
# Create a new provisioning instance
new_provisioning = Provisioning(
qos_profile=create_provisioning.qos_profile,
sink=create_provisioning.sink,
device_id=new_device.id,
sink_credential= \
create_provisioning.sink_credential.credential_type
if create_provisioning.sink_credential
else None
)
# Add the new provisioning to the session
db.add(new_provisioning)
db.commit()
db.refresh(new_provisioning)
return new_provisioning
except SQLAlchemyError as e:
db.rollback()
logger.error(f"Error creating provisioning: {e}")
raise ValueError(f"Error creating provisioning: {e}")
def get_all_provisionings(db: Session, provisioning_id: str) -> Provisioning:
"""
Retrieves all provisioning records from the database.
Args:
db: Database session.
provisioning_id: The ID of the provisioning to query. Although it's
passed, it is not used in the query, and all provisionings are returned.
Returns:
A list of all provisioning records.
"""
return db.query(Provisioning).all()
def update_provisioning_by_id(
db: Session, provisioning_id: str, provisioning_status: str,
provisioning_timestamp: str) -> tuple[Provisioning, Device]:
"""
Updates the status and timestamp of a provisioning record by its ID.
Args:
db: Database session.
provisioning_id: The ID of the provisioning to update.
provisioning_status: The new status for the provisioning.
provisioning_timestamp: The timestamp when the provisioning started.
Returns:
A tuple containing the updated Provisioning object and the associated
Device object.
Raises:
HTTPException: If the fields of the existing device differ during an
update.
"""
provisioning, device = get_provisioning_by_id(db, provisioning_id)
if provisioning:
provisioning.started_at = datetime.fromisoformat(
provisioning_timestamp.replace("Z", "+00:00")
)
provisioning.status = provisioning_status
db.commit()
db.refresh(provisioning)
logger.debug(
f"Updated provisioning with id={provisioning_id} "
f"to status={provisioning_status}"
)
return provisioning, device
def get_provisioning_by_id(
db: Session, provisioning_id: str
) -> tuple[Provisioning, Device]:
"""
Fetch a provisioning by its ID.
Args:
db: Database session.
provisioning_id: The ID of the provisioning.
Returns:
The ProvisioningInfo object or None if not found.
"""
try:
logger.debug(f"Received provisioning ID: {provisioning_id}\n")
# Check if the provisioning exists
provisioning = db.query(Provisioning).filter_by(id=provisioning_id)\
.first()
if provisioning:
device = db.query(Device).filter_by(id=provisioning.device_id)\
.first()
return provisioning, device
else:
logger.debug(f"Provisioning with ID {provisioning_id} not found.\n")
raise HTTPException(
status_code=404,
detail=f"Provisioning with ID {provisioning_id} not found."
)
except SQLAlchemyError as e:
db.rollback()
logger.error(f"Error fetching provisioning by ID: {e}")
raise ValueError(f"Error fetching provisioning by ID: {e}")
def get_provisioning_by_device(
db: Session,
retrieve_provisioning_by_device: RetrieveProvisioningByDevice
) -> tuple[Provisioning, Device]:
"""
Fetch a provisioning by device ID.
Args:
db: Database session.
retrieve_provisioning_by_device: The data needed to retrieve the
provisioning.
Returns:
The ProvisioningInfo object or None if not found.
"""
from fastapi import HTTPException
try:
logger.debug(
f"Received retrieve provisioning by device data: "
f"{retrieve_provisioning_by_device}\n"
)
# Validate if any field to search for is provided
device = retrieve_provisioning_by_device.device
if not (
device.phone_number or
(device.ipv4_address and device.ipv4_address.public_address) or
(device.ipv4_address and device.ipv4_address.private_address) or
(device.ipv4_address and device.ipv4_address.public_port) or
device.ipv6_address or
device.network_access_identifier
):
raise HTTPException(
status_code=400,
detail="No search fields provided to retrieve the device."
)
fields_to_check = retrieve_fields_to_check(device)
# Iterate through the fields to check for the provided values and query the DB
for field, value in fields_to_check:
if value: # Only search if the field has a value
existing_device = db.query(Device).filter_by(**{field: value})\
.first()
logger.debug(
f"Existing device found for field {field}: "
f"{existing_device}"
)
if existing_device: # Stop searching as soon as we find a match
break
# If a device was found, we need to check all fields to ensure they match
if existing_device:
# Compare all fields to check if any field differs
differences = []
for field, value in fields_to_check:
if value and getattr(existing_device, field) != value:
differences.append((field, value))
# If any field differs, raise a conflict (not found)
if differences:
logger.debug(f"Device fields differ: {differences}")
raise HTTPException(
status_code=404,
detail="Device found, but fields differ."
)
logger.debug(
"Device found and fields match, proceeding with provisioning."
)
provisioning = db.query(Provisioning)\
.filter_by(device_id=existing_device.id).first()
if provisioning:
return provisioning, existing_device
else:
logger.debug("Device not found.\n")
raise HTTPException(status_code=404, detail="Device not found.")
except SQLAlchemyError as e:
db.rollback()
logger.error(f"Error fetching provisioning by device: {e}")
raise ValueError(f"Error fetching provisioning by device: {e}")
def delete_provisioning(
db: Session, provisioning_id: str
) -> tuple[Provisioning, Device]:
"""
Deletes a provisioning (marks it as unavailable or removes it).
Args:
db: Database session.
provisioning_id: The ID of the provisioning to delete.
"""
try:
logger.debug(
f"Received to-be-deleted provisioning's id: {provisioning_id}\n"
)
# Check if the provisioning exists
provisioning = db.query(Provisioning).filter_by(id=provisioning_id)\
.first()
if not provisioning:
logger.debug(f"Provisioning with ID {provisioning_id} not found.\n")
raise HTTPException(
status_code=404,
detail=f"Provisioning with ID {provisioning_id} not found."
)
# Check if the device already exists
related_device = db.query(Device).filter_by(id=provisioning.device_id)\
.first()
if related_device:
db.delete(provisioning)
db.commit()
logger.debug(
f"Provisioning with ID {provisioning_id} has been deleted.\n"
)
return provisioning, related_device
else:
logger.debug(
f"Provisioning with ID {provisioning_id} not found.\n"
)
except SQLAlchemyError as e:
db.rollback()
logger.error(f"Error deleting provisioning: {e}")
raise ValueError(f"Error deleting provisioning: {e}")
\ No newline at end of file
QoDProvisioning/QoDProvisioningAPI/API/src/database/db.py 0 → 100644
View file @ 64d0ca5b
# -*- coding: utf-8 -*-
# @Authors:
# Eduardo Santos (eduardosantoshf@av.it.pt)
# Rafael Direito (rdireito@av.it.pt)
# @Organization:
# Instituto de Telecomunicações, Aveiro (ITAv)
# Aveiro, Portugal
# @Date:
# December 2024
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import Session
import os
from config import Config
import logging
logger = Config.setup_logging()
sqlalchemy_logger = logging.getLogger("sqlalchemy.engine")
sqlalchemy_logger.setLevel("WARNING")
for handler in logger.handlers:
sqlalchemy_logger.addHandler(handler)
sqlalchemy_logger.propagate = True
# SQLite database URL
SQLALCHEMY_DATABASE_URL = f"sqlite:///{Config.db_path}"
# Create the SQLAlchemy engine
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# Session and Base for models
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Initialize the database
def init_db():
import database.base_models
Base.metadata.create_all(bind=engine)
# Dependency for database session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
QoDProvisioning/QoDProvisioningAPI/API/src/main.py 0 → 100644
View file @ 64d0ca5b
# -*- coding: utf-8 -*-
# @Authors:
# Eduardo Santos (eduardosantoshf@av.it.pt)
# Rafael Direito (rdireito@av.it.pt)
# @Organization:
# Instituto de Telecomunicações, Aveiro (ITAv)
# Aveiro, Portugal
# @Date:
# December 2024
"""
QoD Provisioning API
The Quality-On-Demand (QoD) Provisioning API offers a programmable interface for developers to request the assignment of a certain QoS Profile to a certain device, indefinitely. This API sets up the configuration in the network so the requested QoS profile is applied to an specified device, at any time while the provisioning is available. The device traffic will be treated with a certain QoS profile by the network whenever the device is connected to the network, until the provisioning is deleted. # Relevant terms and definitions * **QoS profiles and QoS profile labels**: Latency, throughput or priority requirements of the application mapped to relevant QoS profile values. The set of QoS Profiles that a network operator is offering may be retrieved via the `qos-profiles` API (cf. https://github.com/camaraproject/QualityOnDemand/) or will be agreed during the onboarding with the API service provider. * **Identifier for the device**: At least one identifier for the device (user equipment) out of four options: IPv4 address, IPv6 address, Phone number, or Network Access Identifier assigned by the network operator for the device, at the request time. After the provisioning request is accepted, the device may get different IP addresses, but the provisioning will still apply to the device that was identified during the request process. Note: Network Access Identifier is defined for future use and will not be supported with v0.1 of the API. * **Notification URL and token**: Developers may provide a callback URL (`sink`) on which notifications about all status change events (eg. provisioning termination) can be received from the service provider. This is an optional parameter. The notification will be sent as a CloudEvent compliant message. If `sink` is included, it is RECOMMENDED for the client to provide as well the `sinkCredential` property to protect the notification endpoint. In the current version,`sinkCredential.credentialType` MUST be set to `ACCESSTOKEN` if provided. # Resources and Operations overview The API defines four operations: - An operation to setup a new QoD provisioning for a given device. - An operation to get the information about a specific QoD provisioning, identified by its `provisioningId`. - An operation to get the QoD provisioning for a given device. - An operation to terminate a QoD provisioning, identified by its `provisioningId`. # Authorization and Authentication [Camara Security and Interoperability Profile](https://github.com/camaraproject/IdentityAndConsentManagement/blob/main/documentation/CAMARA-Security-Interoperability.md) provides details on how a client requests an access token. Which specific authorization flows are to be used will be determined during onboarding process, happening between the API Client and the Telco Operator exposing the API, taking into account the declared purpose for accessing the API, while also being subject to the prevailing legal framework dictated by local legislation. It is important to remark that in cases where personal user data is processed by the API, and users can exercise their rights through mechanisms such as opt-in and/or opt-out, the use of 3-legged access tokens becomes mandatory. This measure ensures that the API remains in strict compliance with user privacy preferences and regulatory obligations, upholding the principles of transparency and user-centric data control. # Identifying a device from the access token This specification defines the `device` object field as optional in API requests, specifically in cases where the API is accessed using a 3-legged access token, and the device can be uniquely identified by the token. This approach simplifies API usage for API consumers by relying on the device information associated with the access token used to invoke the API. ## Handling of device information: ### Optional device object for 3-legged tokens: - When using a 3-legged access token, the device associated with the access token must be considered as the device for the API request. This means that the device object is not required in the request, and if included it must identify the same device, therefore **it is recommended NOT to include it in these scenarios** to simplify the API usage and avoid additional validations. ### Validation mechanism: - The server will extract the device identification from the access token, if available. - If the API request additionally includes a `device` object when using a 3-legged access token, the API will validate that the device identifier provided matches the one associated with the access token. - If there is a mismatch, the API will respond with a 403 - INVALID_TOKEN_CONTEXT error, indicating that the device information in the request does not match the token. ### Error handling for unidentifiable devices: - If the `device` object is not included in the request and the device information cannot be derived from the 3-legged access token, the server will return a 422 `UNIDENTIFIABLE_DEVICE` error. ### Restrictions for tokens without an associated authenticated identifier: - For scenarios which do not have a single device identifier associated to the token during the authentication flow, e.g. 2-legged access tokens, the `device` object MUST be provided in the API request. This ensures that the device identification is explicit and valid for each API call made with these tokens.
The version of the OpenAPI document: 0.1.0
Generated by OpenAPI Generator (https://openapi-generator.tech)
Do not edit the class manually.
""" # noqa: E501
import asyncio
import json
from fastapi import FastAPI
from sqlalchemy.orm import Session
from routers.qod_provisioning_router import router as QoDProvisioningApiRouter
from routers.osl import router as OSLRouter
from database.db import init_db, get_db
from aux.service_event_manager.service_event_manager import ServiceEventManager
from aux.service_event_manager.camara_results_processor import CamaraResultsProcessor
from config import Config
# Set up logging
logger = Config.setup_logging()
app = FastAPI(
title="QoD Provisioning API",
description=(
"The Quality-On-Demand (QoD) Provisioning API offers a programmable "
"interface for developers to request the assignment of a certain QoS "
"Profile to a certain device, indefinitely.\n\n"
"This API sets up the configuration in the network so the requested QoS profile is applied to a specified device, at any time while the provisioning is available. The device traffic will be treated with a certain QoS profile by the network whenever the device is connected to the network, until the provisioning is deleted.\n\n"
"## Relevant terms and definitions\n\n"
"* **QoS profiles and QoS profile labels**: Latency, throughput or priority requirements of the application mapped to relevant QoS profile values. The set of QoS Profiles that a network operator is offering may be retrieved via the `qos-profiles` API (cf. https://github.com/camaraproject/QualityOnDemand/) or will be agreed during the onboarding with the API service provider.\n\n"
"* **Identifier for the device**: At least one identifier for the device (user equipment) out of four options: IPv4 address, IPv6 address, Phone number, or Network Access Identifier assigned by the network operator for the device, at the request time. After the provisioning request is accepted, the device may get different IP addresses, but the provisioning will still apply to the device that was identified during the request process. Note: Network Access Identifier is defined for future use and will not be supported with v0.1 of the API.\n\n"
"* **Notification URL and token**: Developers may provide a callback URL (`sink`) on which notifications about all status change events (eg. provisioning termination) can be received from the service provider. This is an optional parameter. The notification will be sent as a CloudEvent compliant message. If `sink` is included, it is RECOMMENDED for the client to provide as well the `sinkCredential` property to protect the notification endpoint. In the current version, `sinkCredential.credentialType` MUST be set to `ACCESSTOKEN` if provided.\n\n"
"## Resources and Operations overview\n\n"
"The API defines four operations:\n\n"
"- An operation to setup a new QoD provisioning for a given device.\n\n"
"- An operation to get the information about a specific QoD provisioning, identified by its `provisioningId`.\n\n"
"- An operation to get the QoD provisioning for a given device.\n\n"
"- An operation to terminate a QoD provisioning, identified by its `provisioningId`.\n\n"
"## Authorization and Authentication\n\n"
"[Camara Security and Interoperability Profile](https://github.com/camaraproject/IdentityAndConsentManagement/blob/main/documentation/CAMARA-Security-Interoperability.md) provides details on how a client requests an access token.\n\n"
"Which specific authorization flows are to be used will be determined during onboarding process, happening between the API Client and the Telco Operator exposing the API, taking into account the declared purpose for accessing the API, while also being subject to the prevailing legal framework dictated by local legislation. It is important to remark that in cases where personal user data is processed by the API, and users can exercise their rights through mechanisms such as opt-in and/or opt-out, the use of 3-legged access tokens becomes mandatory. This measure ensures that the API remains in strict compliance with user privacy preferences and regulatory obligations, upholding the principles of transparency and user-centric data control.\n\n"
"## Identifying a device from the access token\n\n"
"This specification defines the `device` object field as optional in API requests, specifically in cases where the API is accessed using a 3-legged access token, and the device can be uniquely identified by the token. This approach simplifies API usage for API consumers by relying on the device information associated with the access token used to invoke the API.\n\n"
"### Handling of device information:\n\n"
"#### Optional device object for 3-legged tokens:\n\n"
"- When using a 3-legged access token, the device associated with the access token must be considered as the device for the API request. This means that the device object is not required in the request, and if included it must identify the same device, therefore **it is recommended NOT to include it in these scenarios** to simplify the API usage and avoid additional validations.\n\n"
"#### Validation mechanism:\n\n"
"- The server will extract the device identification from the access token, if available.\n\n"
"- If the API request additionally includes a `device` object when using a 3-legged access token, the API will validate that the device identifier provided matches the one associated with the access token.\n\n"
"- If there is a mismatch, the API will respond with a 403 - INVALID_TOKEN_CONTEXT error, indicating that the device information in the request does not match the token.\n\n"
"#### Error handling for unidentifiable devices:\n\n"
"- If the `device` object is not included in the request and the device information cannot be derived from the 3-legged access token, the server will return a 422 `UNIDENTIFIABLE_DEVICE` error.\n\n"
"#### Restrictions for tokens without an associated authenticated identifier:\n\n"
"- For scenarios which do not have a single device identifier associated to the token during the authentication flow, e.g. 2-legged access tokens, the `device` object MUST be provided in the API request. This ensures that the device identification is explicit and valid for each API call made with these tokens.\n\n"
),
version="0.1.0",
)
app.include_router(QoDProvisioningApiRouter)
app.include_router(OSLRouter)
@app.on_event("startup")
async def startup_event():
"""
Event triggered when the application starts.
Initializes the database tables.
"""
init_db()
# Initialize the ServiceEventManager and subscribe to OSL events topic
ServiceEventManager.initialize()
ServiceEventManager.subscribe_to_events()
# Initialize the CamaraResultsProcessor with the queue and start processing
camara_processor = CamaraResultsProcessor(
ServiceEventManager.camara_results_queue
)
asyncio.create_task(camara_processor.process_results())
QoDProvisioning/QoDProvisioningAPI/API/src/routers/osl.py 0 → 100644
View file @ 64d0ca5b
# -*- coding: utf-8 -*-
# @Authors:
# Eduardo Santos (eduardosantoshf@av.it.pt)
# Rafael Direito (rdireito@av.it.pt)
# @Organization:
# Instituto de Telecomunicações, Aveiro (ITAv)
# Aveiro, Portugal
# @Date:
# December 2024
from typing import Dict, List # noqa: F401
import importlib
import pkgutil
from fastapi import ( # noqa: F401
APIRouter,
Body,
Cookie,
Depends,
Form,
Header,
HTTPException,
Path,
Query,
Response,
Security,
status,
)
from pydantic import Field, StrictStr
from typing import Any, Optional
from typing_extensions import Annotated
from sqlalchemy.orm import Session
from database.db import get_db
from fastapi import HTTPException, Depends
from schemas.create_provisioning import CreateProvisioning
from schemas.error_info import ErrorInfo
from schemas.provisioning_info import ProvisioningInfo
from schemas.retrieve_provisioning_by_device import RetrieveProvisioningByDevice
from schemas.status import Status
from schemas.status_info import StatusInfo
from database import crud
from aux import mappers
from datetime import datetime
import logging
from aux.service_event_manager.service_event_manager import ServiceEventManager
import json
from config import Config
from aux.constants import Constants
# Set up logging
logger = Config.setup_logging()
router = APIRouter()
@router.get(
"/osl/current-camara-results",
tags=["OSL"],
summary=(
"This endpoint is only used when this service is deployed "
"with OSL. It is used to get a list of the camaraResults "
"processed by the API"
),
response_model_by_alias=True,
status_code=200
)
async def current_camara_results() -> List[ProvisioningInfo]:
return Constants.processed_camara_results
QoDProvisioning/QoDProvisioningAPI/API/src/routers/qod_provisioning_router.py 0 → 100644
View file @ 64d0ca5b
# -*- coding: utf-8 -*-
# @Authors:
# Eduardo Santos (eduardosantoshf@av.it.pt)
# Rafael Direito (rdireito@av.it.pt)
# @Organization:
# Instituto de Telecomunicações, Aveiro (ITAv)
# Aveiro, Portugal
# @Date:
# December 2024
from typing import Dict, List # noqa: F401
import importlib
import pkgutil
from fastapi import ( # noqa: F401
APIRouter,
Body,
Cookie,
Depends,
Form,
Header,
HTTPException,
Path,
Query,
Response,
Security,
status,
)
from schemas.extra_models import TokenModel # noqa: F401
from pydantic import Field, StrictStr
from typing import Any, Optional
from typing_extensions import Annotated
from sqlalchemy.orm import Session
from database.db import get_db
from fastapi import HTTPException, Depends
from schemas.create_provisioning import CreateProvisioning
from schemas.error_info import ErrorInfo
from schemas.provisioning_info import ProvisioningInfo
from schemas.retrieve_provisioning_by_device import RetrieveProvisioningByDevice
from schemas.status import Status
from schemas.status_info import StatusInfo
from database import crud
from aux import mappers
from datetime import datetime
import logging
from aux.service_event_manager.service_event_manager import ServiceEventManager
import json
from config import Config
# Set up logging
logger = Config.setup_logging()
router = APIRouter()
@router.post(
"/device-qos",
responses={
201: {"model": ProvisioningInfo, "description": "Provisioning created"},
400: {"model": ErrorInfo, "description":
"Bad Request with additional errors for implicit notifications"},
401: {"model": ErrorInfo, "description": "Unauthorized"},
403: {"model": ErrorInfo, "description": "Forbidden"},
404: {"model": ErrorInfo, "description": "Not found"},
409: {"model": ErrorInfo, "description": "Provisioning conflict"},
422: {"model": ErrorInfo, "description": "Unprocessable entity"},
429: {"model": ErrorInfo, "description": "Too Many Requests"},
500: {"model": ErrorInfo, "description": "Internal server error"},
503: {"model": ErrorInfo, "description": "Service unavailable"},
},
tags=["QoD Provisioning"],
summary="Sets a new provisioning of QoS for a device",
response_model_by_alias=True,
status_code=201 # Default status code for successful creation
)
async def create_provisioning(
create_provisioning: CreateProvisioning,
x_correlator: Annotated[
Optional[StrictStr],
Field(description="Correlation id for the different services")
] = Header(None, description="Correlation id for the different services"),
db_session: Session = Depends(get_db)
) -> ProvisioningInfo:
try:
# Call the CRUD function to create the provisioning in the database
new_provisioning = crud.create_provisioning(
db_session,
create_provisioning
)
ServiceEventManager.update_service({
"serviceCharacteristic": mappers.map_service_characteristics(
new_provisioning,
"CREATE"
)
})
return ProvisioningInfo(
provisioning_id=new_provisioning.id,
device=mappers.map_device_to_dict(new_provisioning.device),
qos_profile=new_provisioning.qos_profile,
sink=new_provisioning.sink,
sink_credential={
"credential_type": new_provisioning.sink_credential
},
started_at=datetime.utcnow(),
status=new_provisioning.status,
status_info=new_provisioning.status_info
)
except HTTPException:
raise
except Exception as e:
# If an error occurs, roll back and raise an HTTPException
raise HTTPException(status_code=500, detail=str(e))
@router.delete(
"/device-qos/{provisioningId}",
responses={
204: {"description": "Provisioning deleted"},
202: {
"model": ProvisioningInfo,
"description": (
"Deletion request accepted to be processed. "
"It applies for an async deletion process. "
"`status` in the response will be `AVAILABLE` "
"with `statusInfo` set to `DELETE_REQUESTED`."
)
},
400: {"model": ErrorInfo, "description": "Bad Request"},
401: {"model": ErrorInfo, "description": "Unauthorized"},
403: {"model": ErrorInfo, "description": "Forbidden"},
404: {"model": ErrorInfo, "description": "Not found"},
429: {"model": ErrorInfo, "description": "Too Many Requests"},
500: {"model": ErrorInfo, "description": "Internal server error"},
503: {"model": ErrorInfo, "description": "Service unavailable"},
},
tags=["QoD Provisioning"],
summary="Deletes a QoD provisioning",
response_model_by_alias=True,
)
async def delete_provisioning(
provisioningId: Annotated[
StrictStr,
Field(description=(
"Provisioning ID that was obtained from the createProvision "
"operation"
))
] = Path(..., description=(
"Provisioning ID that was obtained from the createProvision "
"operation"
)),
x_correlator: Annotated[
Optional[StrictStr],
Field(description="Correlation id for the different services")
] = Header(None, description="Correlation id for the different services"),
db_session: Session = Depends(get_db),
) -> ProvisioningInfo:
"""
Release resources related to QoS provisioning.
If the notification callback is provided and the provisioning status was
`AVAILABLE`, when the deletion is completed, the client will
receive in addition to the response a
`PROVISIONING_STATUS_CHANGED` event with - `status`
as `UNAVAILABLE` and - `statusInfo` as
`DELETE_REQUESTED` There will be no notification event if the
`status` was already `UNAVAILABLE`.
**NOTES:** - The access token may be either 2-legged or 3-legged. -
If a 3-legged access token is used, the end user (and device) associated
with the QoD provisioning must also be associated with the access token. -
The QoD provisioning must have been created by the same API client given in
the access token.
"""
try:
# Call the CRUD function to create the provisioning in the database
provisioning, related_device = crud.delete_provisioning(
db_session, provisioningId
)
ServiceEventManager.update_service({
"serviceCharacteristic": mappers.map_service_characteristics(
provisioning,
"DELETE"
)
})
deleted_provisioning = ProvisioningInfo(
provisioning_id=str(provisioning.id),
device=mappers.map_device_to_dict(related_device),
qos_profile=provisioning.qos_profile,
sink=provisioning.sink,
sink_credential={
"credential_type": provisioning.sink_credential
},
started_at=provisioning.started_at,
status=provisioning.status,
status_info=StatusInfo.DELETE_REQUESTED
)
return deleted_provisioning
except HTTPException:
# Allow 404 and other HTTPExceptions to propagate without modification
raise
except Exception as e:
# If an error occurs, roll back and raise an HTTPException
raise HTTPException(status_code=500, detail=str(e))
@router.get(
"/device-qos/{provisioningId}",
responses={
200: {
"model": ProvisioningInfo,
"description": "Returns information about certain provisioning"
},
400: {"model": ErrorInfo, "description": "Bad Request"},
401: {"model": ErrorInfo, "description": "Unauthorized"},
403: {"model": ErrorInfo, "description": "Forbidden"},
404: {"model": ErrorInfo, "description": "Not found"},
429: {"model": ErrorInfo, "description": "Too Many Requests"},
500: {"model": ErrorInfo, "description": "Internal server error"},
503: {"model": ErrorInfo, "description": "Service unavailable"},
},
tags=["QoD Provisioning"],
summary="Get QoD provisioning information",
response_model_by_alias=True,
)
async def get_provisioning_by_id(
provisioningId: Annotated[
StrictStr,
Field(description=(
"Provisioning ID that was obtained from the createProvision "
"operation"
))
] = Path(..., description=(
"Provisioning ID that was obtained from the createProvision operation"
)),
x_correlator: Annotated[
Optional[StrictStr],
Field(description="Correlation id for the different services")
] = Header(None, description="Correlation id for the different services"),
db_session: Session = Depends(get_db)
) -> ProvisioningInfo:
try:
# Call the CRUD function to create the provisioning in the database
provisioning, device = crud.get_provisioning_by_id(
db_session, provisioningId
)
if provisioning.status_info:
provisioning_status_info = provisioning.status_info
else:
provisioning_status_info = None
retrieved_provisioning = ProvisioningInfo(
provisioning_id=str(provisioning.id),
device=mappers.map_device_to_dict(device),
qos_profile=provisioning.qos_profile,
sink=provisioning.sink,
sink_credential={
"credential_type": provisioning.sink_credential
},
started_at=provisioning.started_at,
status=provisioning.status,
status_info=provisioning_status_info
)
return retrieved_provisioning
except HTTPException:
# Allow 404 and other HTTPExceptions to propagate without modification
raise
except Exception as e:
# If an error occurs, roll back and raise an HTTPException
raise HTTPException(status_code=500, detail=str(e))
@router.post(
"/retrieve-device-qos",
responses={
200: {
"model": ProvisioningInfo,
"description": (
"Returns information about QoS provisioning for the device."
)
},
400: {"model": ErrorInfo, "description": "Bad Request"},
401: {"model": ErrorInfo, "description": "Unauthorized"},
403: {"model": ErrorInfo, "description": "Forbidden"},
404: {"model": ErrorInfo, "description": "Not found"},
422: {"model": ErrorInfo, "description": "Unprocessable entity"},
429: {"model": ErrorInfo, "description": "Too Many Requests"},
500: {"model": ErrorInfo, "description": "Internal server error"},
503: {"model": ErrorInfo, "description": "Service unavailable"},
},
tags=["QoD Provisioning"],
summary="Gets the QoD provisioning for a device",
response_model_by_alias=True,
)
async def retrieve_provisioning_by_device(
retrieve_provisioning_by_device: Annotated[
RetrieveProvisioningByDevice,
Field(description="Parameters to retrieve a provisioning by device")
] = Body(
None, description="Parameters to retrieve a provisioning by device"
),
x_correlator: Annotated[
Optional[StrictStr],
Field(description="Correlation id for the different services")
] = Header(None, description="Correlation id for the different services"),
db_session: Session = Depends(get_db)
) -> ProvisioningInfo:
"""
Retrieves the QoD provisioning for a device. **NOTES:** - The access token
may be either 2-legged or 3-legged. - If a 3-legged access token is used,
the end user (and device) associated with the QoD provisioning must also be
associated with the access token. In this case it is recommended NOT to
include the `device` parameter in the request (see \"
Handling of device information\" within the API description for
details). - If a 2-legged access token is used, the device parameter must
be provided and identify a device. - The QoD provisioning must have been
created by the same API client given in the access token. - If no
provisioning is found for the device, an error response 404 is returned with
code \"NOT_FOUND\".
"""
try:
# Call the CRUD function to create the provisioning in the database
provisioning, existing_device = crud.get_provisioning_by_device(
db_session, retrieve_provisioning_by_device
)
if provisioning.status_info:
provisioning_status_info = provisioning.status_info
else:
provisioning_status_info = None
device_provisioning_info = ProvisioningInfo(
provisioning_id=str(provisioning.id),
device=mappers.map_device_to_dict(existing_device),
qos_profile=provisioning.qos_profile,
sink=provisioning.sink,
sink_credential={
"credential_type": provisioning.sink_credential
},
started_at=provisioning.started_at,
status=provisioning.status,
status_info=provisioning_status_info
)
return device_provisioning_info
except HTTPException:
# Allow 404 and other HTTPExceptions to propagate without modification
raise
except Exception as e:
# If an error occurs, roll back and raise an HTTPException
raise HTTPException(status_code=500, detail=str(e))
QoDProvisioning/QoDProvisioningAPI/API/src/schemas/access_token_credential.py 0 → 100644
View file @ 64d0ca5b
# coding: utf-8
"""
QoD Provisioning API
The Quality-On-Demand (QoD) Provisioning API offers a programmable interface for developers to request the assignment of a certain QoS Profile to a certain device, indefinitely. This API sets up the configuration in the network so the requested QoS profile is applied to an specified device, at any time while the provisioning is available. The device traffic will be treated with a certain QoS profile by the network whenever the device is connected to the network, until the provisioning is deleted. # Relevant terms and definitions * **QoS profiles and QoS profile labels**: Latency, throughput or priority requirements of the application mapped to relevant QoS profile values. The set of QoS Profiles that a network operator is offering may be retrieved via the `qos-profiles` API (cf. https://github.com/camaraproject/QualityOnDemand/) or will be agreed during the onboarding with the API service provider. * **Identifier for the device**: At least one identifier for the device (user equipment) out of four options: IPv4 address, IPv6 address, Phone number, or Network Access Identifier assigned by the network operator for the device, at the request time. After the provisioning request is accepted, the device may get different IP addresses, but the provisioning will still apply to the device that was identified during the request process. Note: Network Access Identifier is defined for future use and will not be supported with v0.1 of the API. * **Notification URL and token**: Developers may provide a callback URL (`sink`) on which notifications about all status change events (eg. provisioning termination) can be received from the service provider. This is an optional parameter. The notification will be sent as a CloudEvent compliant message. If `sink` is included, it is RECOMMENDED for the client to provide as well the `sinkCredential` property to protect the notification endpoint. In the current version,`sinkCredential.credentialType` MUST be set to `ACCESSTOKEN` if provided. # Resources and Operations overview The API defines four operations: - An operation to setup a new QoD provisioning for a given device. - An operation to get the information about a specific QoD provisioning, identified by its `provisioningId`. - An operation to get the QoD provisioning for a given device. - An operation to terminate a QoD provisioning, identified by its `provisioningId`. # Authorization and Authentication [Camara Security and Interoperability Profile](https://github.com/camaraproject/IdentityAndConsentManagement/blob/main/documentation/CAMARA-Security-Interoperability.md) provides details on how a client requests an access token. Which specific authorization flows are to be used will be determined during onboarding process, happening between the API Client and the Telco Operator exposing the API, taking into account the declared purpose for accessing the API, while also being subject to the prevailing legal framework dictated by local legislation. It is important to remark that in cases where personal user data is processed by the API, and users can exercise their rights through mechanisms such as opt-in and/or opt-out, the use of 3-legged access tokens becomes mandatory. This measure ensures that the API remains in strict compliance with user privacy preferences and regulatory obligations, upholding the principles of transparency and user-centric data control. # Identifying a device from the access token This specification defines the `device` object field as optional in API requests, specifically in cases where the API is accessed using a 3-legged access token, and the device can be uniquely identified by the token. This approach simplifies API usage for API consumers by relying on the device information associated with the access token used to invoke the API. ## Handling of device information: ### Optional device object for 3-legged tokens: - When using a 3-legged access token, the device associated with the access token must be considered as the device for the API request. This means that the device object is not required in the request, and if included it must identify the same device, therefore **it is recommended NOT to include it in these scenarios** to simplify the API usage and avoid additional validations. ### Validation mechanism: - The server will extract the device identification from the access token, if available. - If the API request additionally includes a `device` object when using a 3-legged access token, the API will validate that the device identifier provided matches the one associated with the access token. - If there is a mismatch, the API will respond with a 403 - INVALID_TOKEN_CONTEXT error, indicating that the device information in the request does not match the token. ### Error handling for unidentifiable devices: - If the `device` object is not included in the request and the device information cannot be derived from the 3-legged access token, the server will return a 422 `UNIDENTIFIABLE_DEVICE` error. ### Restrictions for tokens without an associated authenticated identifier: - For scenarios which do not have a single device identifier associated to the token during the authentication flow, e.g. 2-legged access tokens, the `device` object MUST be provided in the API request. This ensures that the device identification is explicit and valid for each API call made with these tokens.
The version of the OpenAPI document: 0.1.0
Generated by OpenAPI Generator (https://openapi-generator.tech)
Do not edit the class manually.
""" # noqa: E501
from __future__ import annotations
import pprint
import re # noqa: F401
import json
from datetime import datetime
from pydantic import ConfigDict, Field, StrictStr, field_validator
from typing import Any, ClassVar, Dict, List
from schemas.sink_credential import SinkCredential
try:
from typing import Self
except ImportError:
from typing_extensions import Self
class AccessTokenCredential(SinkCredential):
"""
An access token credential.
""" # noqa: E501
credential_type: StrictStr = Field(description="The type of the credential.", alias="credentialType")
access_token: StrictStr = Field(description="REQUIRED. An access token is a previously acquired token granting access to the target resource.", alias="accessToken")
access_token_expires_utc: datetime = Field(description="REQUIRED. An absolute UTC instant at which the token shall be considered expired.", alias="accessTokenExpiresUtc")
access_token_type: StrictStr = Field(description="REQUIRED. Type of the access token (See [OAuth 2.0](https://tools.ietf.org/html/rfc6749#section-7.1)).", alias="accessTokenType")
__properties: ClassVar[List[str]] = ["credentialType", "accessToken", "accessTokenExpiresUtc", "accessTokenType"]
@field_validator('credential_type')
def credential_type_validate_enum(cls, value):
"""Validates the enum"""
if value not in ('PLAIN', 'ACCESSTOKEN', 'REFRESHTOKEN',):
raise ValueError("must be one of enum values ('PLAIN', 'ACCESSTOKEN', 'REFRESHTOKEN')")
return value
@field_validator('access_token_type')
def access_token_type_validate_enum(cls, value):
"""Validates the enum"""
if value not in ('bearer',):
raise ValueError("must be one of enum values ('bearer')")
return value
model_config = {
"populate_by_name": True,
"validate_assignment": True,
"protected_namespaces": (),
}
def to_str(self) -> str:
"""Returns the string representation of the model using alias"""
return pprint.pformat(self.model_dump(by_alias=True))
def to_json(self) -> str:
"""Returns the JSON representation of the model using alias"""
# TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead
return json.dumps(self.to_dict())
@classmethod
def from_json(cls, json_str: str) -> Self:
"""Create an instance of AccessTokenCredential from a JSON string"""
return cls.from_dict(json.loads(json_str))
def to_dict(self) -> Dict[str, Any]:
"""Return the dictionary representation of the model using alias.
This has the following differences from calling pydantic's
`self.model_dump(by_alias=True)`:
* `None` is only added to the output dict for nullable fields that
were set at model initialization. Other fields with value `None`
are ignored.
"""
_dict = self.model_dump(
by_alias=True,
exclude={
},
exclude_none=True,
)
return _dict
@classmethod
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of AccessTokenCredential from a dict"""
if obj is None:
return None
if not isinstance(obj, dict):
return cls.model_validate(obj)
_obj = cls.model_validate({
"credentialType": obj.get("credentialType"),
"accessToken": obj.get("accessToken"),
"accessTokenExpiresUtc": obj.get("accessTokenExpiresUtc"),
"accessTokenType": obj.get("accessTokenType")
})
return _obj
QoDProvisioning/QoDProvisioningAPI/API/src/schemas/base_provisioning_info.py 0 → 100644
View file @ 64d0ca5b
# coding: utf-8
"""
QoD Provisioning API
The Quality-On-Demand (QoD) Provisioning API offers a programmable interface for developers to request the assignment of a certain QoS Profile to a certain device, indefinitely. This API sets up the configuration in the network so the requested QoS profile is applied to an specified device, at any time while the provisioning is available. The device traffic will be treated with a certain QoS profile by the network whenever the device is connected to the network, until the provisioning is deleted. # Relevant terms and definitions * **QoS profiles and QoS profile labels**: Latency, throughput or priority requirements of the application mapped to relevant QoS profile values. The set of QoS Profiles that a network operator is offering may be retrieved via the `qos-profiles` API (cf. https://github.com/camaraproject/QualityOnDemand/) or will be agreed during the onboarding with the API service provider. * **Identifier for the device**: At least one identifier for the device (user equipment) out of four options: IPv4 address, IPv6 address, Phone number, or Network Access Identifier assigned by the network operator for the device, at the request time. After the provisioning request is accepted, the device may get different IP addresses, but the provisioning will still apply to the device that was identified during the request process. Note: Network Access Identifier is defined for future use and will not be supported with v0.1 of the API. * **Notification URL and token**: Developers may provide a callback URL (`sink`) on which notifications about all status change events (eg. provisioning termination) can be received from the service provider. This is an optional parameter. The notification will be sent as a CloudEvent compliant message. If `sink` is included, it is RECOMMENDED for the client to provide as well the `sinkCredential` property to protect the notification endpoint. In the current version,`sinkCredential.credentialType` MUST be set to `ACCESSTOKEN` if provided. # Resources and Operations overview The API defines four operations: - An operation to setup a new QoD provisioning for a given device. - An operation to get the information about a specific QoD provisioning, identified by its `provisioningId`. - An operation to get the QoD provisioning for a given device. - An operation to terminate a QoD provisioning, identified by its `provisioningId`. # Authorization and Authentication [Camara Security and Interoperability Profile](https://github.com/camaraproject/IdentityAndConsentManagement/blob/main/documentation/CAMARA-Security-Interoperability.md) provides details on how a client requests an access token. Which specific authorization flows are to be used will be determined during onboarding process, happening between the API Client and the Telco Operator exposing the API, taking into account the declared purpose for accessing the API, while also being subject to the prevailing legal framework dictated by local legislation. It is important to remark that in cases where personal user data is processed by the API, and users can exercise their rights through mechanisms such as opt-in and/or opt-out, the use of 3-legged access tokens becomes mandatory. This measure ensures that the API remains in strict compliance with user privacy preferences and regulatory obligations, upholding the principles of transparency and user-centric data control. # Identifying a device from the access token This specification defines the `device` object field as optional in API requests, specifically in cases where the API is accessed using a 3-legged access token, and the device can be uniquely identified by the token. This approach simplifies API usage for API consumers by relying on the device information associated with the access token used to invoke the API. ## Handling of device information: ### Optional device object for 3-legged tokens: - When using a 3-legged access token, the device associated with the access token must be considered as the device for the API request. This means that the device object is not required in the request, and if included it must identify the same device, therefore **it is recommended NOT to include it in these scenarios** to simplify the API usage and avoid additional validations. ### Validation mechanism: - The server will extract the device identification from the access token, if available. - If the API request additionally includes a `device` object when using a 3-legged access token, the API will validate that the device identifier provided matches the one associated with the access token. - If there is a mismatch, the API will respond with a 403 - INVALID_TOKEN_CONTEXT error, indicating that the device information in the request does not match the token. ### Error handling for unidentifiable devices: - If the `device` object is not included in the request and the device information cannot be derived from the 3-legged access token, the server will return a 422 `UNIDENTIFIABLE_DEVICE` error. ### Restrictions for tokens without an associated authenticated identifier: - For scenarios which do not have a single device identifier associated to the token during the authentication flow, e.g. 2-legged access tokens, the `device` object MUST be provided in the API request. This ensures that the device identification is explicit and valid for each API call made with these tokens.
The version of the OpenAPI document: 0.1.0
Generated by OpenAPI Generator (https://openapi-generator.tech)
Do not edit the class manually.
""" # noqa: E501
from __future__ import annotations
import pprint
import re # noqa: F401
import json
from pydantic import BaseModel, ConfigDict, Field, StrictStr, field_validator
from typing import Any, ClassVar, Dict, List, Optional
from typing_extensions import Annotated
from schemas.device import Device
from schemas.sink_credential import SinkCredential
try:
from typing import Self
except ImportError:
from typing_extensions import Self
class BaseProvisioningInfo(BaseModel):
"""
Common attributes of a QoD provisioning
""" # noqa: E501
device: Optional[Device] = None
qos_profile: Annotated[str, Field(min_length=3, strict=True, max_length=256)] = Field(description="A unique name for identifying a specific QoS profile. This may follow different formats depending on the service providers implementation. Some options addresses: - A UUID style string - Support for predefined profiles QOS_S, QOS_M, QOS_L, and QOS_E - A searchable descriptive name The set of QoS Profiles that an operator is offering can be retrieved by means of the [QoS Profile API](link TBC). ", alias="qosProfile")
sink: Optional[StrictStr] = Field(default=None, description="The address to which events shall be delivered, using the HTTP protocol.")
sink_credential: Optional[SinkCredential] = Field(default=None, alias="sinkCredential")
__properties: ClassVar[List[str]] = ["device", "qosProfile", "sink", "sinkCredential"]
@field_validator('qos_profile')
def qos_profile_validate_regular_expression(cls, value):
"""Validates the regular expression"""
if not re.match(r"^[a-zA-Z0-9_.-]+$", value):
raise ValueError(r"must validate the regular expression /^[a-zA-Z0-9_.-]+$/")
return value
model_config = {
"populate_by_name": True,
"validate_assignment": True,
"protected_namespaces": (),
}
def to_str(self) -> str:
"""Returns the string representation of the model using alias"""
return pprint.pformat(self.model_dump(by_alias=True))
def to_json(self) -> str:
"""Returns the JSON representation of the model using alias"""
# TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead
return json.dumps(self.to_dict())
@classmethod
def from_json(cls, json_str: str) -> Self:
"""Create an instance of BaseProvisioningInfo from a JSON string"""
return cls.from_dict(json.loads(json_str))
def to_dict(self) -> Dict[str, Any]:
"""Return the dictionary representation of the model using alias.
This has the following differences from calling pydantic's
`self.model_dump(by_alias=True)`:
* `None` is only added to the output dict for nullable fields that
were set at model initialization. Other fields with value `None`
are ignored.
"""
_dict = self.model_dump(
by_alias=True,
exclude={
},
exclude_none=True,
)
# override the default output from pydantic by calling `to_dict()` of device
if self.device:
_dict['device'] = self.device.to_dict()
# override the default output from pydantic by calling `to_dict()` of sink_credential
if self.sink_credential:
_dict['sinkCredential'] = self.sink_credential.to_dict()
return _dict
@classmethod
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of BaseProvisioningInfo from a dict"""
if obj is None:
return None
if not isinstance(obj, dict):
return cls.model_validate(obj)
_obj = cls.model_validate({
"device": Device.from_dict(obj.get("device")) if obj.get("device") is not None else None,
"qosProfile": obj.get("qosProfile"),
"sink": obj.get("sink"),
"sinkCredential": SinkCredential.from_dict(obj.get("sinkCredential")) if obj.get("sinkCredential") is not None else None
})
return _obj
QoDProvisioning/QoDProvisioningAPI/API/src/schemas/cloud_event.py 0 → 100644
View file @ 64d0ca5b
# coding: utf-8
"""
QoD Provisioning API
The Quality-On-Demand (QoD) Provisioning API offers a programmable interface for developers to request the assignment of a certain QoS Profile to a certain device, indefinitely. This API sets up the configuration in the network so the requested QoS profile is applied to an specified device, at any time while the provisioning is available. The device traffic will be treated with a certain QoS profile by the network whenever the device is connected to the network, until the provisioning is deleted. # Relevant terms and definitions * **QoS profiles and QoS profile labels**: Latency, throughput or priority requirements of the application mapped to relevant QoS profile values. The set of QoS Profiles that a network operator is offering may be retrieved via the `qos-profiles` API (cf. https://github.com/camaraproject/QualityOnDemand/) or will be agreed during the onboarding with the API service provider. * **Identifier for the device**: At least one identifier for the device (user equipment) out of four options: IPv4 address, IPv6 address, Phone number, or Network Access Identifier assigned by the network operator for the device, at the request time. After the provisioning request is accepted, the device may get different IP addresses, but the provisioning will still apply to the device that was identified during the request process. Note: Network Access Identifier is defined for future use and will not be supported with v0.1 of the API. * **Notification URL and token**: Developers may provide a callback URL (`sink`) on which notifications about all status change events (eg. provisioning termination) can be received from the service provider. This is an optional parameter. The notification will be sent as a CloudEvent compliant message. If `sink` is included, it is RECOMMENDED for the client to provide as well the `sinkCredential` property to protect the notification endpoint. In the current version,`sinkCredential.credentialType` MUST be set to `ACCESSTOKEN` if provided. # Resources and Operations overview The API defines four operations: - An operation to setup a new QoD provisioning for a given device. - An operation to get the information about a specific QoD provisioning, identified by its `provisioningId`. - An operation to get the QoD provisioning for a given device. - An operation to terminate a QoD provisioning, identified by its `provisioningId`. # Authorization and Authentication [Camara Security and Interoperability Profile](https://github.com/camaraproject/IdentityAndConsentManagement/blob/main/documentation/CAMARA-Security-Interoperability.md) provides details on how a client requests an access token. Which specific authorization flows are to be used will be determined during onboarding process, happening between the API Client and the Telco Operator exposing the API, taking into account the declared purpose for accessing the API, while also being subject to the prevailing legal framework dictated by local legislation. It is important to remark that in cases where personal user data is processed by the API, and users can exercise their rights through mechanisms such as opt-in and/or opt-out, the use of 3-legged access tokens becomes mandatory. This measure ensures that the API remains in strict compliance with user privacy preferences and regulatory obligations, upholding the principles of transparency and user-centric data control. # Identifying a device from the access token This specification defines the `device` object field as optional in API requests, specifically in cases where the API is accessed using a 3-legged access token, and the device can be uniquely identified by the token. This approach simplifies API usage for API consumers by relying on the device information associated with the access token used to invoke the API. ## Handling of device information: ### Optional device object for 3-legged tokens: - When using a 3-legged access token, the device associated with the access token must be considered as the device for the API request. This means that the device object is not required in the request, and if included it must identify the same device, therefore **it is recommended NOT to include it in these scenarios** to simplify the API usage and avoid additional validations. ### Validation mechanism: - The server will extract the device identification from the access token, if available. - If the API request additionally includes a `device` object when using a 3-legged access token, the API will validate that the device identifier provided matches the one associated with the access token. - If there is a mismatch, the API will respond with a 403 - INVALID_TOKEN_CONTEXT error, indicating that the device information in the request does not match the token. ### Error handling for unidentifiable devices: - If the `device` object is not included in the request and the device information cannot be derived from the 3-legged access token, the server will return a 422 `UNIDENTIFIABLE_DEVICE` error. ### Restrictions for tokens without an associated authenticated identifier: - For scenarios which do not have a single device identifier associated to the token during the authentication flow, e.g. 2-legged access tokens, the `device` object MUST be provided in the API request. This ensures that the device identification is explicit and valid for each API call made with these tokens.
The version of the OpenAPI document: 0.1.0
Generated by OpenAPI Generator (https://openapi-generator.tech)
Do not edit the class manually.
""" # noqa: E501
from __future__ import annotations
import pprint
import re # noqa: F401
import json
from datetime import datetime
from importlib import import_module
from pydantic import BaseModel, ConfigDict, Field, StrictStr, field_validator
from typing import Any, ClassVar, Dict, List, Optional, Union
try:
from typing import Self
except ImportError:
from typing_extensions import Self
class CloudEvent(BaseModel):
"""
Event compliant with the CloudEvents specification
""" # noqa: E501
id: StrictStr = Field(description="Identifier of this event, that must be unique in the source context.")
source: StrictStr = Field(description="Identifies the context in which an event happened in the specific Provider Implementation.")
type: StrictStr = Field(description="The type of the event.")
specversion: StrictStr = Field(description="Version of the specification to which this event conforms (must be 1.0 if it conforms to cloudevents 1.0.2 version)")
datacontenttype: Optional[StrictStr] = Field(default=None, description="media-type that describes the event payload encoding, must be \"application/json\" for CAMARA APIs")
data: Optional[Dict[str, Any]] = Field(default=None, description="Event notification details payload, which depends on the event type")
time: datetime = Field(description="Timestamp of when the occurrence happened. It must follow RFC 3339 ")
__properties: ClassVar[List[str]] = ["id", "source", "type", "specversion", "datacontenttype", "data", "time"]
@field_validator('type')
def type_validate_enum(cls, value):
"""Validates the enum"""
if value not in ('org.camaraproject.qod-provisioning.v0.status-changed',):
raise ValueError("must be one of enum values ('org.camaraproject.qod-provisioning.v0.status-changed')")
return value
@field_validator('specversion')
def specversion_validate_enum(cls, value):
"""Validates the enum"""
if value not in ('1.0',):
raise ValueError("must be one of enum values ('1.0')")
return value
@field_validator('datacontenttype')
def datacontenttype_validate_enum(cls, value):
"""Validates the enum"""
if value is None:
return value
if value not in ('application/json',):
raise ValueError("must be one of enum values ('application/json')")
return value
model_config = {
"populate_by_name": True,
"validate_assignment": True,
"protected_namespaces": (),
}
# JSON field name that stores the object type
__discriminator_property_name: ClassVar[List[str]] = 'type'
# discriminator mappings
__discriminator_value_class_map: ClassVar[Dict[str, str]] = {
'org.camaraproject.qod-provisioning.v0.status-changed': 'EventStatusChanged'
}
@classmethod
def get_discriminator_value(cls, obj: Dict) -> str:
"""Returns the discriminator value (object type) of the data"""
discriminator_value = obj[cls.__discriminator_property_name]
if discriminator_value:
return cls.__discriminator_value_class_map.get(discriminator_value)
else:
return None
def to_str(self) -> str:
"""Returns the string representation of the model using alias"""
return pprint.pformat(self.model_dump(by_alias=True))
def to_json(self) -> str:
"""Returns the JSON representation of the model using alias"""
# TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead
return json.dumps(self.to_dict())
@classmethod
def from_json(cls, json_str: str) -> Union[Self]:
"""Create an instance of CloudEvent from a JSON string"""
return cls.from_dict(json.loads(json_str))
def to_dict(self) -> Dict[str, Any]:
"""Return the dictionary representation of the model using alias.
This has the following differences from calling pydantic's
`self.model_dump(by_alias=True)`:
* `None` is only added to the output dict for nullable fields that
were set at model initialization. Other fields with value `None`
are ignored.
"""
_dict = self.model_dump(
by_alias=True,
exclude={
},
exclude_none=True,
)
return _dict
@classmethod
def from_dict(cls, obj: Dict) -> Union[Self]:
"""Create an instance of CloudEvent from a dict"""
# look up the object type based on discriminator mapping
object_type = cls.get_discriminator_value(obj)
if object_type:
klass = globals()[object_type]
return klass.from_dict(obj)
else:
raise ValueError("CloudEvent failed to lookup discriminator value from " +
json.dumps(obj) + ". Discriminator property name: " + cls.__discriminator_property_name +
", mapping: " + json.dumps(cls.__discriminator_value_class_map))
QoDProvisioning/QoDProvisioningAPI/API/src/schemas/create_provisioning.py 0 → 100644
View file @ 64d0ca5b
# -*- coding: utf-8 -*-
# @Author: Eduardo Santos
# @Date: 2024-11-28 10:13:05
# @Last Modified by: Eduardo Santos
# @Last Modified time: 2024-11-28 17:29:05
# coding: utf-8
"""
QoD Provisioning API
The Quality-On-Demand (QoD) Provisioning API offers a programmable interface for developers to request the assignment of a certain QoS Profile to a certain device, indefinitely. This API sets up the configuration in the network so the requested QoS profile is applied to an specified device, at any time while the provisioning is available. The device traffic will be treated with a certain QoS profile by the network whenever the device is connected to the network, until the provisioning is deleted. # Relevant terms and definitions * **QoS profiles and QoS profile labels**: Latency, throughput or priority requirements of the application mapped to relevant QoS profile values. The set of QoS Profiles that a network operator is offering may be retrieved via the `qos-profiles` API (cf. https://github.com/camaraproject/QualityOnDemand/) or will be agreed during the onboarding with the API service provider. * **Identifier for the device**: At least one identifier for the device (user equipment) out of four options: IPv4 address, IPv6 address, Phone number, or Network Access Identifier assigned by the network operator for the device, at the request time. After the provisioning request is accepted, the device may get different IP addresses, but the provisioning will still apply to the device that was identified during the request process. Note: Network Access Identifier is defined for future use and will not be supported with v0.1 of the API. * **Notification URL and token**: Developers may provide a callback URL (`sink`) on which notifications about all status change events (eg. provisioning termination) can be received from the service provider. This is an optional parameter. The notification will be sent as a CloudEvent compliant message. If `sink` is included, it is RECOMMENDED for the client to provide as well the `sinkCredential` property to protect the notification endpoint. In the current version,`sinkCredential.credentialType` MUST be set to `ACCESSTOKEN` if provided. # Resources and Operations overview The API defines four operations: - An operation to setup a new QoD provisioning for a given device. - An operation to get the information about a specific QoD provisioning, identified by its `provisioningId`. - An operation to get the QoD provisioning for a given device. - An operation to terminate a QoD provisioning, identified by its `provisioningId`. # Authorization and Authentication [Camara Security and Interoperability Profile](https://github.com/camaraproject/IdentityAndConsentManagement/blob/main/documentation/CAMARA-Security-Interoperability.md) provides details on how a client requests an access token. Which specific authorization flows are to be used will be determined during onboarding process, happening between the API Client and the Telco Operator exposing the API, taking into account the declared purpose for accessing the API, while also being subject to the prevailing legal framework dictated by local legislation. It is important to remark that in cases where personal user data is processed by the API, and users can exercise their rights through mechanisms such as opt-in and/or opt-out, the use of 3-legged access tokens becomes mandatory. This measure ensures that the API remains in strict compliance with user privacy preferences and regulatory obligations, upholding the principles of transparency and user-centric data control. # Identifying a device from the access token This specification defines the `device` object field as optional in API requests, specifically in cases where the API is accessed using a 3-legged access token, and the device can be uniquely identified by the token. This approach simplifies API usage for API consumers by relying on the device information associated with the access token used to invoke the API. ## Handling of device information: ### Optional device object for 3-legged tokens: - When using a 3-legged access token, the device associated with the access token must be considered as the device for the API request. This means that the device object is not required in the request, and if included it must identify the same device, therefore **it is recommended NOT to include it in these scenarios** to simplify the API usage and avoid additional validations. ### Validation mechanism: - The server will extract the device identification from the access token, if available. - If the API request additionally includes a `device` object when using a 3-legged access token, the API will validate that the device identifier provided matches the one associated with the access token. - If there is a mismatch, the API will respond with a 403 - INVALID_TOKEN_CONTEXT error, indicating that the device information in the request does not match the token. ### Error handling for unidentifiable devices: - If the `device` object is not included in the request and the device information cannot be derived from the 3-legged access token, the server will return a 422 `UNIDENTIFIABLE_DEVICE` error. ### Restrictions for tokens without an associated authenticated identifier: - For scenarios which do not have a single device identifier associated to the token during the authentication flow, e.g. 2-legged access tokens, the `device` object MUST be provided in the API request. This ensures that the device identification is explicit and valid for each API call made with these tokens.
The version of the OpenAPI document: 0.1.0
Generated by OpenAPI Generator (https://openapi-generator.tech)
Do not edit the class manually.
""" # noqa: E501
from __future__ import annotations
import pprint
import re # noqa: F401
import json
from pydantic import BaseModel, ConfigDict, Field, StrictStr, field_validator
from typing import Any, ClassVar, Dict, List, Optional
from typing_extensions import Annotated
from schemas.device import Device
from schemas.sink_credential import SinkCredential
try:
from typing import Self
except ImportError:
from typing_extensions import Self
class CreateProvisioning(BaseModel):
"""
Attributes to request a new QoD provisioning
""" # noqa: E501
device: Optional[Device] = None
qos_profile: Annotated[str, Field(min_length=3, strict=True, max_length=256)] = Field(description="A unique name for identifying a specific QoS profile. This may follow different formats depending on the service providers implementation. Some options addresses: - A UUID style string - Support for predefined profiles QOS_S, QOS_M, QOS_L, and QOS_E - A searchable descriptive name The set of QoS Profiles that an operator is offering can be retrieved by means of the [QoS Profile API](link TBC). ", alias="qosProfile")
sink: Optional[StrictStr] = Field(default=None, description="The address to which events shall be delivered, using the HTTP protocol.")
sink_credential: Optional[SinkCredential] = Field(default=None, alias="sinkCredential")
__properties: ClassVar[List[str]] = ["device", "qosProfile", "sink", "sinkCredential"]
@field_validator('qos_profile')
def qos_profile_validate_regular_expression(cls, value):
"""Validates the regular expression"""
if not re.match(r"^[a-zA-Z0-9_.-]+$", value):
raise ValueError(r"must validate the regular expression /^[a-zA-Z0-9_.-]+$/")
return value
model_config = {
"populate_by_name": True,
"validate_assignment": True,
"protected_namespaces": (),
}
def to_str(self) -> str:
"""Returns the string representation of the model using alias"""
return pprint.pformat(self.model_dump(by_alias=True))
def to_json(self) -> str:
"""Returns the JSON representation of the model using alias"""
# TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead
return json.dumps(self.to_dict())
@classmethod
def from_json(cls, json_str: str) -> Self:
"""Create an instance of CreateProvisioning from a JSON string"""
return cls.from_dict(json.loads(json_str))
def to_dict(self) -> Dict[str, Any]:
"""Return the dictionary representation of the model using alias.
This has the following differences from calling pydantic's
`self.model_dump(by_alias=True)`:
* `None` is only added to the output dict for nullable fields that
were set at model initialization. Other fields with value `None`
are ignored.
"""
_dict = self.model_dump(
by_alias=True,
exclude={
},
exclude_none=True,
)
# override the default output from pydantic by calling `to_dict()` of device
if self.device:
_dict['device'] = self.device.to_dict()
# override the default output from pydantic by calling `to_dict()` of sink_credential
if self.sink_credential:
_dict['sinkCredential'] = self.sink_credential.to_dict()
return _dict
@classmethod
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of CreateProvisioning from a dict"""
if obj is None:
return None
if not isinstance(obj, dict):
return cls.model_validate(obj)
_obj = cls.model_validate({
"device": Device.from_dict(obj.get("device")) if obj.get("device") is not None else None,
"qosProfile": obj.get("qosProfile"),
"sink": obj.get("sink"),
"sinkCredential": SinkCredential.from_dict(obj.get("sinkCredential")) if obj.get("sinkCredential") is not None else None
})
return _obj
QoDProvisioning/QoDProvisioningAPI/API/src/schemas/device.py 0 → 100644
View file @ 64d0ca5b
This diff is collapsed.
QoDProvisioning/QoDProvisioningAPI/API/src/schemas/extra_models.py 0 → 100644
View file @ 64d0ca5b
# coding: utf-8
from pydantic import BaseModel
class TokenModel(BaseModel):
"""Defines a token model."""
sub: str