Microservice Discovery Through a Manifest File

version: 1
metadata:
  name: disputes-service-v1
  externalId: disputes-service-v1
  description: |
    A microservice responsible for payment disputes.
    This service handles payment transaction disputes and is an integral part of our payment ecosystem.
  type: Backend  
  repository:
    url: https://example.com
    status: active
    visibility: private

  applications:
    - factSheetId: fa787383-7233-4896-8fad-c1f1bef30dd2

  tags:
    - tagGroupName: Domain
      tagNames:
        - Payments

    - tagGroupName: Location
      tagNames:
        - AWS-EU1
        - AWS-EU2

  teams:
    - factSheetId: 63eda74c-57f7-4768-b1c9-3b2813b11504
    - factSheetId: afd1ee0f-095b-4c68-88f6-d3628070ce18

  resources:
    - name: Disputes Process Flow
      type: documentation
      url: https://myorg.atlassian.net/wiki/spaces/disputes
      description: Disputes process flow and diagrams

import logging
from pathlib import Path
import yaml
import requests
import json
import os

logging.basicConfig(level=logging.INFO)

# Request timeout
TIMEOUT = 20

# API token and Subdomain are set as env variables.
# It is adviced not to hard code sensitive information in your code.
LEANIX_API_TOKEN = os.getenv("LEANIX_API_TOKEN")
LEANIX_SUBDOMAIN = os.getenv("LEANIX_SUBDOMAIN")
LEANIX_FQDN = f"https://{LEANIX_SUBDOMAIN}.leanix.net/services"
LEANIX_MANIFEST_FILE = os.getenv("LEANIX_MANIFEST_FILE", "leanix.yaml")

# OAuth2 URL to request the access token.
LEANIX_OAUTH2_URL = f"{LEANIX_FQDN}/mtm/v1/oauth2/token"

# GraphQL API
LEANIX_GRAPHQL_API = f"{LEANIX_FQDN}/pathfinder/v1/graphql"

# SBOM API endpoint
LEANIX_SBOM_API = f"{LEANIX_FQDN}/technology-discovery/v1/microservices"

# Github Related
GITHUB_SERVER_URL = os.getenv("GITHUB_SERVER_URL")
GITHUB_REPOSITORY = os.getenv("GITHUB_REPOSITORY")


def obtain_access_token() -> str:
    """Obtains a LeanIX Access token using the Technical User generated
    API secret.

    Returns:
        str: The LeanIX Access Token
    """
    if not LEANIX_API_TOKEN:
        raise Exception("A valid token is required")
    response = requests.post(
        LEANIX_OAUTH2_URL,
        auth=("apitoken", LEANIX_API_TOKEN),
        data={"grant_type": "client_credentials"},
    )
    response.raise_for_status()
    return response.json().get("access_token")


def _parse_manifest_file() -> dict:
    """Parses the Manifest file and generates the payload for the
    API request for the LeanIX Microservices API.

    Returns:
        dict: The payload for the API request
    """
    manifest_data = dict()
    with open(LEANIX_MANIFEST_FILE, "r") as file:
        try:
            logging.info(f"Parsing manifest file: {file.name}")
            manifest_data = yaml.safe_load(file)
        except yaml.YAMLError as exc:
            logging.error(f"Failed to load Manifest file: {exc}")

    if not manifest_data:
        logging.info("No manifest entries found")
        return manifest_data
    manifest_microservice = manifest_data.get("metadata", {})
    return manifest_microservice


def microservice_exists(external_id: str) -> dict:
    """
    This function sends a GraphQL query to check if a microservice with the given external ID exists.
    It makes the request, and returns the microservice's data if it exists, or an empty dictionary if it does not.

    Args:
        external_id (str): The external ID of the microservice to check.

    Returns:
        dict: A dictionary containing the microservice's data if it exists. If the microservice does not exist,
        an empty dictionary is returned.
    """
    query = """
    query CheckMicroservice($externalIds: [String!]) {
        factSheets(externalIds: $externalIds) {
            edges {
                node {
                    ... on Application {
                        id
                        name
                        relApplicationToOwningOrganization {
                            edges {
                                node {
                                    factSheet {
                                        id
                                        name
                                    }
                                }
                            }
                        }
                        tags {
                            id
                            name
                        }
                    }
                }
            }
        }
    }
    """
    leanix_external_id = f"technologyDiscoveryId/{external_id}"
    variables = {"externalIds": [leanix_external_id]}
    resp = make_request(query, variables)
    response_data = resp.json().get("data", {}).get("factSheets", {}).get("edges", [])
    fact_sheet = dict()
    if response_data:
        fact_sheet = response_data[0].get("node", {})
    return fact_sheet


def generate_owning_teams(microservice_manifest: dict) -> list[dict] | list:
    """
    This function generates a list of dictionaries representing the owning teams of a microservice.
    Each dictionary represents an operation to add a new relationship between the microservice and
    its owning team. The operation is defined with fields 'op', 'path', and 'value'. The 'value'
    field contains the fact sheet ID of the owning team in a JSON format.

    Args:
        microservice_manifest (dict): A dictionary containing the microservice manifest data.
        It includes a 'teams' key which is a list of dictionaries, each containing details
        about a team including its 'factSheetId'.

    Returns:
        list[dict]: A list of dictionaries, each representing an operation to add a new relationship
        between the microservice and its owning team. If no teams are found in the manifest data,
        an empty list is returned.
    """
    teams = list()
    for id, team in enumerate(microservice_manifest.get("teams", []), start=1):
        fact_sheet_id = team.get("factSheetId")
        if fact_sheet_id:
            teams.append(
                {
                    "op": "add",
                    "path": f"/relApplicationToOwningOrganization/new_{id}",
                    "value": json.dumps({"factSheetId": fact_sheet_id}),
                }
            )
    return teams


def generate_graphql_payload(microservice_manifest: dict) -> dict:
    """
    This function generates a payload dictionary for a microservice based on the provided manifest data.
    The payload includes key information about the microservice such as its type and external ID,
    which are extracted from the manifest dictionary. This payload can be used to register or update
    the microservice in a system.

    Args:
        microservice_manifest (dict): A dictionary containing the manifest data for the microservice.

    Returns:
        dict: A dictionary containing the payload data for the microservice.
    """
    tags = list()
    for tagGroup in microservice_manifest.get("tags", []):
        for tagName in tagGroup.get("tagNames", []):
            tags.append({"tagName": tagName})

    variables = {
        "patches": [
            {"op": "replace", "path": "/category", "value": "microservice"},
            {
                "op": "replace",
                "path": "/description",
                "value": microservice_manifest.get("description"),
            },
            {
                "op": "replace",
                "path": "/technologyDiscoveryId",
                "value": json.dumps(
                    {"externalId": microservice_manifest.get("externalId")}
                ),
            },
            {
                "op": "replace",
                "path": "/lxRepositoryUrl",
                "value": f"{GITHUB_SERVER_URL}/{GITHUB_REPOSITORY}",
            },
            {"op": "replace", "path": "/lxRepositoryStatus", "value": "active"},
            {"op": "replace", "path": "/lxRepositoryVisibility", "value": "public"},
        ],
    }
    if len(tags):
        variables["patches"].append(
            {"op": "replace", "path": "/tags", "value": json.dumps(tags)}
        )
    return variables


def make_request(query: str, variables: dict | None) -> requests.Response:
    """
    Sends a GraphQL request to the LeanIX API.

    This function sends a POST request to the LeanIX API with the given GraphQL query and variables.
    It uses the LEANIX_ACCESS_TOKEN environment variable for authentication.

    Args:
        query (str): The GraphQL query to send in the request.
        variables (dict, optional): A dictionary containing the variables for the GraphQL query.

    Returns:
        requests.Response: The response from the LeanIX API. If the request fails, an exception is raised.
    """
    url = f"{LEANIX_GRAPHQL_API}"
    # Fetch the access token and set the Authorization Header
    auth_header = f'Bearer {os.environ.get("LEANIX_ACCESS_TOKEN")}'
    # Provide the headers
    headers = {
        "Authorization": auth_header,
    }
    response = requests.post(
        headers=headers, url=url, json={"query": query, "variables": variables}
    )
    response.raise_for_status()
    return response


def create_or_update_micro_services(manifest_payload: dict):
    """
    Creates or updates the LeanIX Microservice Fact Sheet based on the provided manifest file.

    This function checks if a microservice with the given external ID exists. If it does, the microservice is updated.
    If it does not exist, a new microservice is created. After the microservice is created or updated,
    the function triggers the registration of the relevant SBOM file with LeanIX.

    Args:
        manifest_payload (dict): A dictionary containing the manifest data for the microservice.
        This data includes the microservices name and external ID.
    """
    external_id = manifest_payload.get("externalId")
    microservice_name = manifest_payload.get("name")
    fact_sheet_id = None
    fact_sheet = dict()
    if not external_id:
        logging.error("Could not determine external id for microservice")
        return
    fact_sheet = microservice_exists(external_id)
    fact_sheet_id = fact_sheet.get("id")
    if not fact_sheet_id:
        logging.info(f"Microservice {microservice_name} does not exist")
        fact_sheet = create_microservice(manifest_payload)
        if not fact_sheet:
            logging.error(f"Failed to create microservice {microservice_name}")
            return
        fact_sheet_id = fact_sheet.get("id")
        if fact_sheet_id:
            logging.info(
                f"Created microservice {microservice_name}, fact sheet id: {fact_sheet_id}"
            )
    else:
        logging.info(
            f"Microservice {microservice_name} exists, fact sheet id: {fact_sheet_id}"
        )
        fact_sheet = update_microservice(manifest_payload, fact_sheet)
        if not fact_sheet:
            logging.error(f"Failed to update microservice {microservice_name}")
            return
        if fact_sheet:
            logging.info(f"Successfully updated microservice: {microservice_name}")
    if fact_sheet_id:
        register_sboms(fact_sheet_id)


def create_microservice(manifest_data: dict) -> dict:
    """
    This function sends a GraphQL mutation to create a new microservice with the given manifest data.
    It generates the necessary payload, makes the request, and returns the created microservice's data.
    If any errors occur during the creation process, an empty dictionary is returned.

    Args:
        manifest_data (dict): A dictionary containing the manifest data for the microservice.
        This data includes the microservice's name, type, and owning teams' fact sheet IDs.

    Returns:
        dict: A dictionary containing the created microservice's data. If any errors occur during
        the creation process, an empty dictionary is returned.
    """
    query = """
    mutation CreateMicroservice($input: BaseFactSheetInput!, $patches: [Patch]) {
        createFactSheet(input: $input, patches: $patches) {
            factSheet {
                id
                name
                type
                category
                ... on Application {
                    technologyDiscoveryId {
                        externalId
                        externalUrl
                        externalVersion
                    }
                    lxRepositoryUrl,
                    lxRepositoryStatus,
                    lxRepositoryVisibility
                    relApplicationToOwningOrganization {
                        edges {
                            node {
                                factSheet {
                                    name
                                    id
                                }
                            }
                        }
                    }
                    tags {
                        id
                        name
                    }
                }
            }
        }
    }
    """
    variables = generate_graphql_payload(manifest_data)
    variables["input"] = {"name": manifest_data.get("name"), "type": "Application"}
    teams = generate_owning_teams(manifest_data)
    variables["patches"].extend(teams)
    resp = make_request(query, variables)
    errors = resp.json().get("errors", [])
    if errors:
        logging.error(
            f"Errors encountered while creating microservice, errrors: {errors}"
        )
        return {}
    return resp.json().get("data", {}).get("createFactSheet", {}).get("factSheet", {})


def update_microservice(manifest_data: dict, fact_sheet: dict) -> dict:
    """
    This function sends a GraphQL mutation to update a microservice with the given manifest data.
    It generates the necessary payload, makes the request, and returns the updated microservice's data.
    If any errors occur during the update process, an error message is logged and an empty dictionary is returned.

    Args:
        manifest_data (dict): A dictionary containing the manifest data for the microservice.
        This data includes the microservice's name, type, and owning teams' fact sheet IDs.

        fact_sheet (dict): A dictionary containing the current fact sheet data of the microservice to be updated.
        It includes the microservice's ID.

    Returns:
        dict: A dictionary containing the updated microservice's data. If any errors occur during
        the update process, an empty dictionary is returned.
    """
    query = """
    mutation UpdateMicroservice($ID: ID!, $patches: [Patch]!) {
        updateFactSheet(id: $ID, patches: $patches) {
            factSheet {
                id
                name
                type
                category
                ... on Application {
                    technologyDiscoveryId {
                        externalId
                        externalUrl
                        externalVersion
                    }
                    lxRepositoryUrl,
                    lxRepositoryStatus,
                    lxRepositoryVisibility
                    relApplicationToOwningOrganization {
                        edges {
                            node {
                                factSheet {
                                    name
                                    id
                                }
                            }
                        }
                    }
                    tags {
                        id
                        name
                    }
                }
            }
        }
    }
    """
    variables = generate_graphql_payload(manifest_data)
    variables["ID"] = fact_sheet.get("id")
    resp = make_request(query, variables)
    errors = resp.json().get("errors", [])
    if errors:
        logging.error(
            f"Errors encountered while updating microservice: {manifest_data.get('name')}, errors: {errors}"
        )
        return {}
    return resp.json().get("data", {}).get("updateFactSheet", {}).get("factSheet", {})


def register_sboms(factsheet_id: str):
    """
    Registers the Software Bill of Materials (SBOM) file with LeanIX.

    This function enables improved understanding of the dependency landscape of your microservices.
    The SBOM provides comprehensive details about software components, their relationships, and
    attributes, which are crucial for managing, securing, and licensing your open-source software.
    By registering the SBOM with LeanIX, these details can be effectively managed and tracked.

    Args:
        factsheet_id (str): The unique identifier of the microservice fact sheet. This ID is used
        to associate the SBOM with the corresponding microservice in LeanIX.

    Returns:
        None
    """
    sbom_path = Path("sbom.json")
    if not sbom_path.exists():
        logging.warning("No sbom file found")
        return

    url = f"{LEANIX_SBOM_API}/{factsheet_id}/sboms"
    sbom_contents = dict()
    logging.info(
        f"Processing sbom file: {sbom_path.name} for Fact Sheet: {factsheet_id}"
    )
    with sbom_path.open("rb") as f:
        sbom_contents = f.read()

    request_payload = {
        "sbom": (
            sbom_path.name,
            sbom_contents,
            "application/json",
        )
    }
    logging.debug(f"Populated payload for SBOM: {sbom_path.name}")
    # Fetch the access token and set the Authorization Header
    auth_header = f'Bearer {os.environ.get("LEANIX_ACCESS_TOKEN")}'
    # Provide the headers
    # NOTE: Don't set the content type, `requests` should handle this.
    headers = {
        "Authorization": auth_header,
    }
    logging.info(f"Sending sbom ingestion request for Fact Sheet: {factsheet_id}")
    response = requests.post(
        url, headers=headers, files=request_payload, timeout=TIMEOUT
    )
    response.raise_for_status()
    logging.info(f"Successfully submited sbom request for Fact Sheet: {factsheet_id}")


def main():
    """LeanIX helper to parse the manifest file create or update a microservice
    and register the relevant dependencies.
    """
    manifest_data = _parse_manifest_file()
    create_or_update_micro_services(manifest_data)


if __name__ == "__main__":
    # Set the access token as an environment variable
    os.environ["LEANIX_ACCESS_TOKEN"] = obtain_access_token()
    main()