RabbitMQ integration

.github/workflows/test.yaml

@@ -8,5 +8,5 @@ jobs:
     uses: canonical/operator-workflows/.github/workflows/test.yaml@main
     secrets: inherit
     with:
-      self-hosted-runner: true
+      self-hosted-runner: false
       self-hosted-runner-label: "edge"

.trivyignore

@@ -1,8 +1,6 @@
-# stdlib golang: net/netip: Unexpected behavior from Is methods for IPv4-mapped IPv6 addresses
-CVE-2024-24790
-# CVE usr/bin/pebble (gobinary)
-CVE-2023-45288
 # ignore CVE introduced by python3-gunicorn
 CVE-2022-40897
 # pypa/setuptools: Remote code execution via download
 CVE-2024-6345
+# pebble: Calling Decoder.Decode on a message which contains deeply nested structures can cause a panic due to stack exhaustion
+CVE-2024-34156

examples/flask/charmcraft.yaml

@@ -115,6 +115,11 @@ requires:
     interface: saml
     optional: True
     limit: 1
+  rabbitmq:
+    interface: rabbitmq
+    optional: True
+    limit: 1
+
 resources:
   flask-app-image:
     description: flask application image.

examples/flask/test_rock/app.py

@@ -8,6 +8,7 @@
 
 import boto3
 import botocore.config
+import pika
 import psycopg
 import pymongo
 import pymongo.database
@@ -73,6 +74,35 @@ def get_redis_database() -> redis.Redis | None:
     return g.redis_db
 
 
+def get_rabbitmq_connection() -> pika.BlockingConnection | None:
+    """Get rabbitmq connection."""
+    if "rabbitmq" not in g:
+        if "RABBITMQ_HOSTNAME" in os.environ:
+            username = os.environ["RABBITMQ_USERNAME"]
+            password = os.environ["RABBITMQ_PASSWORD"]
+            hostname = os.environ["RABBITMQ_HOSTNAME"]
+            vhost = os.environ["RABBITMQ_VHOST"]
+            port = os.environ["RABBITMQ_PORT"]
+            credentials = pika.PlainCredentials(username, password)
+            parameters = pika.ConnectionParameters(hostname, port, vhost, credentials)
+            g.rabbitmq = pika.BlockingConnection(parameters)
+        else:
+            return None
+    return g.rabbitmq
+
+
+def get_rabbitmq_connection_from_uri() -> pika.BlockingConnection | None:
+    """Get rabbitmq connection from uri."""
+    if "rabbitmq_from_uri" not in g:
+        if "RABBITMQ_CONNECT_STRING" in os.environ:
+            uri = os.environ["RABBITMQ_CONNECT_STRING"]
+            parameters = pika.URLParameters(uri)
+            g.rabbitmq_from_uri = pika.BlockingConnection(parameters)
+        else:
+            return None
+    return g.rabbitmq_from_uri
+
+
 def get_boto3_client():
     if "boto3_client" not in g:
         if "S3_ACCESS_KEY" in os.environ:
@@ -113,6 +143,12 @@ def teardown_database(_):
     boto3_client = g.pop("boto3_client", None)
     if boto3_client is not None:
         boto3_client.close()
+    rabbitmq = g.pop("rabbitmq", None)
+    if rabbitmq is not None:
+        rabbitmq.close()
+    rabbitmq_from_uri = g.pop("rabbitmq_from_uri", None)
+    if rabbitmq_from_uri is not None:
+        rabbitmq_from_uri.close()
 
 
 @app.route("/")
@@ -187,6 +223,32 @@ def redis_status():
     return "FAIL"
 
 
+@app.route("/rabbitmq/send")
+def rabbitmq_send():
+    """Send a message to "charm" queue."""
+    if connection := get_rabbitmq_connection():
+        channel = connection.channel()
+        channel.queue_declare(queue="charm")
+        channel.basic_publish(exchange="", routing_key="charm", body="SUCCESS")
+        return "SUCCESS"
+    return "FAIL"
+
+
+@app.route("/rabbitmq/receive")
+def rabbitmq_receive():
+    """Receive a message from "charm" queue in blocking form."""
+    if connection := get_rabbitmq_connection_from_uri():
+        channel = connection.channel()
+        method_frame, _header_frame, body = channel.basic_get("charm")
+        if method_frame:
+            channel.basic_ack(method_frame.delivery_tag)
+            if body == b"SUCCESS":
+                return "SUCCESS"
+            return "FAIL. INCORRECT MESSAGE."
+        return "FAIL. NO MESSAGE."
+    return "FAIL. NO CONNECTION."
+
+
 @app.route("/env")
 def get_env():
     """Return environment variables"""

examples/flask/test_rock/requirements.txt

@@ -6,3 +6,4 @@ psycopg[binary]
 pymongo
 redis[hiredis]
 boto3
+pika

paas_app_charmer/app.py

@@ -219,10 +219,10 @@ def map_integrations_to_env(integrations: IntegrationsState, prefix: str = "") -
     """
     env = {}
     if integrations.redis_uri:
-        redis_envvars = _db_url_to_env_variables("redis", integrations.redis_uri)
+        redis_envvars = _db_url_to_env_variables("REDIS", integrations.redis_uri)
         env.update(redis_envvars)
     for interface_name, uri in integrations.databases_uris.items():
-        interface_envvars = _db_url_to_env_variables(interface_name, uri)
+        interface_envvars = _db_url_to_env_variables(interface_name.upper(), uri)
         env.update(interface_envvars)
 
     if integrations.s3_parameters:
@@ -259,46 +259,85 @@ def map_integrations_to_env(integrations: IntegrationsState, prefix: str = "") -
             if v is not None
         )
 
+    if integrations.rabbitmq_uri:
+        rabbitmq_envvars = _rabbitmq_uri_to_env_variables("RABBITMQ", integrations.rabbitmq_uri)
+        env.update(rabbitmq_envvars)
+
     return {prefix + k: v for k, v in env.items()}
 
 
-def _db_url_to_env_variables(base_name: str, url: str) -> dict[str, str]:
+def _db_url_to_env_variables(prefix: str, url: str) -> dict[str, str]:
     """Convert a database url to environment variables.
 
     Args:
-      base_name: name of the database.
+      prefix: prefix for the environment variables
       url: url of the database
 
     Return:
       All environment variables, that is, the connection string,
       all components as returned from urllib.parse and the
       database name extracted from the path
     """
+    prefix = prefix + "_DB"
+    envvars = _url_env_vars(prefix, url)
+    parsed_url = urllib.parse.urlparse(url)
+
+    # database name is usually parsed this way.
+    db_name = parsed_url.path.removeprefix("/") if parsed_url.path else None
+    if db_name is not None:
+        envvars[f"{prefix}_NAME"] = db_name
+    return envvars
+
+
+def _rabbitmq_uri_to_env_variables(prefix: str, url: str) -> dict[str, str]:
+    """Convert a rabbitmq uri to environment variables.
+
+    Args:
+      prefix: prefix for the environment variables
+      url: url of rabbitmq
+
+    Return:
+      All environment variables, that is, the connection string,
+      all components as returned from urllib.parse and the
+      rabbitmq vhost extracted from the path
+    """
+    envvars = _url_env_vars(prefix, url)
+    parsed_url = urllib.parse.urlparse(url)
+    if len(parsed_url.path) > 1:
+        envvars[f"{prefix}_VHOST"] = urllib.parse.unquote(parsed_url.path.split("/")[1])
+    return envvars
+
+
+def _url_env_vars(prefix: str, url: str) -> dict[str, str]:
+    """Convert a url to environment variables using parts from urllib.parse.urlparse.
+
+    Args:
+      prefix: prefix for the environment variables
+      url: url of the database
+
+    Return:
+      All environment variables, that is, the connection string and
+      all components as returned from urllib.parse
+    """
     if not url:
         return {}
 
-    base_name = base_name.upper()
     envvars: dict[str, str | None] = {}
-    envvars[f"{base_name}_DB_CONNECT_STRING"] = url
+    envvars[f"{prefix}_CONNECT_STRING"] = url
 
     parsed_url = urllib.parse.urlparse(url)
 
     # All components of urlparse, using the same convention for default values.
     # See: https://docs.python.org/3/library/urllib.parse.html#url-parsing
-    envvars[f"{base_name}_DB_SCHEME"] = parsed_url.scheme
-    envvars[f"{base_name}_DB_NETLOC"] = parsed_url.netloc
-    envvars[f"{base_name}_DB_PATH"] = parsed_url.path
-    envvars[f"{base_name}_DB_PARAMS"] = parsed_url.params
-    envvars[f"{base_name}_DB_QUERY"] = parsed_url.query
-    envvars[f"{base_name}_DB_FRAGMENT"] = parsed_url.fragment
-    envvars[f"{base_name}_DB_USERNAME"] = parsed_url.username
-    envvars[f"{base_name}_DB_PASSWORD"] = parsed_url.password
-    envvars[f"{base_name}_DB_HOSTNAME"] = parsed_url.hostname
-    envvars[f"{base_name}_DB_PORT"] = str(parsed_url.port) if parsed_url.port is not None else None
-
-    # database name is usually parsed this way.
-    envvars[f"{base_name}_DB_NAME"] = (
-        parsed_url.path.removeprefix("/") if parsed_url.path else None
-    )
+    envvars[f"{prefix}_SCHEME"] = parsed_url.scheme
+    envvars[f"{prefix}_NETLOC"] = parsed_url.netloc
+    envvars[f"{prefix}_PATH"] = parsed_url.path
+    envvars[f"{prefix}_PARAMS"] = parsed_url.params
+    envvars[f"{prefix}_QUERY"] = parsed_url.query
+    envvars[f"{prefix}_FRAGMENT"] = parsed_url.fragment
+    envvars[f"{prefix}_USERNAME"] = parsed_url.username
+    envvars[f"{prefix}_PASSWORD"] = parsed_url.password
+    envvars[f"{prefix}_HOSTNAME"] = parsed_url.hostname
+    envvars[f"{prefix}_PORT"] = str(parsed_url.port) if parsed_url.port is not None else None
 
     return {k: v for k, v in envvars.items() if v is not None}

paas_app_charmer/charm.py

@@ -19,6 +19,7 @@
 from paas_app_charmer.databases import make_database_requirers
 from paas_app_charmer.exceptions import CharmConfigInvalidError
 from paas_app_charmer.observability import Observability
+from paas_app_charmer.rabbitmq import RabbitMQRequires
 from paas_app_charmer.secret_storage import KeySecretStorage
 from paas_app_charmer.utils import build_validation_error_message
 
@@ -101,6 +102,20 @@ def __init__(self, framework: ops.Framework, framework_name: str) -> None:
         else:
             self._saml = None
 
+        self._rabbitmq: RabbitMQRequires | None
+        if "rabbitmq" in requires and requires["rabbitmq"].interface_name == "rabbitmq":
+            self._rabbitmq = RabbitMQRequires(
+                self,
+                "rabbitmq",
+                username=self.app.name,
+                vhost="/",
+            )
+            self.framework.observe(self._rabbitmq.on.connected, self._on_rabbitmq_connected)
+            self.framework.observe(self._rabbitmq.on.ready, self._on_rabbitmq_ready)
+            self.framework.observe(self._rabbitmq.on.departed, self._on_rabbitmq_departed)
+        else:
+            self._rabbitmq = None
+
         self._database_migration = DatabaseMigration(
             container=self.unit.get_container(self._workload_config.container_name),
             state_dir=self._workload_config.state_dir,
@@ -245,7 +260,8 @@ def is_ready(self) -> bool:
 
         return True
 
-    def _missing_required_integrations(self, charm_state: CharmState) -> list[str]:
+    # Pending to refactor all integrations
+    def _missing_required_integrations(self, charm_state: CharmState) -> list[str]:  # noqa: C901
         """Get list of missing integrations that are required.
 
         Args:
@@ -272,6 +288,9 @@ def _missing_required_integrations(self, charm_state: CharmState) -> list[str]:
         if self._saml and not charm_state.integrations.saml_parameters:
             if not requires["saml"].optional:
                 missing_integrations.append("saml")
+        if self._rabbitmq and not charm_state.integrations.rabbitmq_uri:
+            if not requires["rabbitmq"].optional:
+                missing_integrations.append("rabbitmq")
         return missing_integrations
 
     def restart(self) -> None:
@@ -319,6 +338,7 @@ def _create_charm_state(self) -> CharmState:
             redis_uri=self._redis.url if self._redis is not None else None,
             s3_connection_info=self._s3.get_s3_connection_info() if self._s3 else None,
             saml_relation_data=saml_relation_data,
+            rabbitmq_uri=self._rabbitmq.rabbitmq_uri() if self._rabbitmq else None,
             base_url=self._base_url,
         )
 
@@ -418,3 +438,18 @@ def _on_ingress_ready(self, _: ops.HookEvent) -> None:
     def _on_pebble_ready(self, _: ops.PebbleReadyEvent) -> None:
         """Handle the pebble-ready event."""
         self.restart()
+
+    @block_if_invalid_config
+    def _on_rabbitmq_connected(self, _event: ops.HookEvent) -> None:
+        """Handle rabbitmq connected event."""
+        self.restart()
+
+    @block_if_invalid_config
+    def _on_rabbitmq_ready(self, _event: ops.HookEvent) -> None:
+        """Handle rabbitmq ready event."""
+        self.restart()
+
+    @block_if_invalid_config
+    def _on_rabbitmq_departed(self, _event: ops.HookEvent) -> None:
+        """Handle rabbitmq departed event."""
+        self.restart()

paas_app_charmer/charm_state.py

@@ -88,6 +88,7 @@ def from_charm(  # pylint: disable=too-many-arguments
         redis_uri: str | None = None,
         s3_connection_info: dict[str, str] | None = None,
         saml_relation_data: typing.MutableMapping[str, str] | None = None,
+        rabbitmq_uri: str | None = None,
         base_url: str | None = None,
     ) -> "CharmState":
         """Initialize a new instance of the CharmState class from the associated charm.
@@ -101,6 +102,7 @@ def from_charm(  # pylint: disable=too-many-arguments
             redis_uri: The redis uri provided by the redis charm.
             s3_connection_info: Connection info from S3 lib.
             saml_relation_data: Relation data from the SAML app.
+            rabbitmq_uri: RabbitMQ uri.
             base_url: Base URL for the service.
 
         Return:
@@ -120,6 +122,7 @@ def from_charm(  # pylint: disable=too-many-arguments
             database_requirers=database_requirers,
             s3_connection_info=s3_connection_info,
             saml_relation_data=saml_relation_data,
+            rabbitmq_uri=rabbitmq_uri,
         )
         return cls(
             framework=framework,
@@ -205,20 +208,24 @@ class IntegrationsState:
         databases_uris: Map from interface_name to the database uri.
         s3_parameters: S3 parameters.
         saml_parameters: SAML parameters.
+        rabbitmq_uri: RabbitMQ uri.
     """
 
     redis_uri: str | None = None
     databases_uris: dict[str, str] = field(default_factory=dict)
     s3_parameters: "S3Parameters | None" = None
     saml_parameters: "SamlParameters | None" = None
+    rabbitmq_uri: str | None = None
 
+    # This dataclass combines all the integrations, so it is reasonable that they stay together.
     @classmethod
-    def build(
+    def build(  # pylint: disable=too-many-arguments
         cls,
         redis_uri: str | None,
         database_requirers: dict[str, DatabaseRequires],
         s3_connection_info: dict[str, str] | None,
         saml_relation_data: typing.MutableMapping[str, str] | None = None,
+        rabbitmq_uri: str | None = None,
     ) -> "IntegrationsState":
         """Initialize a new instance of the IntegrationsState class.
 
@@ -229,6 +236,7 @@ def build(
             database_requirers: All database requirers object declared by the charm.
             s3_connection_info: S3 connection info from S3 lib.
             saml_relation_data: Saml relation data from saml lib.
+            rabbitmq_uri: RabbitMQ uri.
 
         Return:
             The IntegrationsState instance created.
@@ -276,6 +284,7 @@ def build(
             },
             s3_parameters=s3_parameters,
             saml_parameters=saml_parameters,
+            rabbitmq_uri=rabbitmq_uri,
         )