Add OCI detection via IMDS metadata server

cloud_detect/providers/oci_provider.py

@@ -1,6 +1,10 @@
+import asyncio
+import contextlib
 import logging
 from pathlib import Path
 
+import aiohttp
+
 from .provider import AbstractProvider
 
 
@@ -12,17 +16,42 @@ class OCIProvider(AbstractProvider):
 
     def __init__(self, logger=None):
         self.logger = logger or logging.getLogger(__name__)
+        self.metadata_url = 'http://169.254.169.254/opc/v1/instance/'
+        self.metadata_url_v2 = 'http://169.254.169.254/opc/v2/instance/'
+        self.headers = {'Authorization': 'Bearer Oracle'}
         self.vendor_file = '/sys/class/dmi/id/chassis_asset_tag'
 
     async def identify(self):
         """
             Tries to identify OCI using all the implemented options
         """
         self.logger.info('Try to identify OCI')
-        return self.check_vendor_file()
+        return self.check_vendor_file() or await self.check_metadata_server()
+
+    async def _get_metadata_v2(self):
+        with contextlib.suppress(BaseException):
+            return await self._get_metadata(url=self.metadata_url_v2, headers=self.headers)
+        return False
 
-    def check_metadata_server(self):
-        raise NotImplementedError
+    async def _get_metadata(self, url=None, headers=None):
+        with contextlib.suppress(BaseException):
+            async with aiohttp.ClientSession() as session:
+                async with session.get(url or self.metadata_url, headers=headers) as response:
+                    response = await response.json(content_type=None)
+                    if str(response.get('id', '')).startswith('ocid1.instance.'):
+                        return True
+        return False
+
+    async def check_metadata_server(self):
+        """
+            Tries to identify OCI via metadata server
+        """
+        self.logger.debug('Checking OCI metadata')
+        results = await asyncio.gather(
+            self._get_metadata(),
+            self._get_metadata_v2(),
+        )
+        return any(results)
 
     def check_vendor_file(self):
         """

tests/oci_provider_test.py

@@ -1,3 +1,5 @@
+import pytest
+
 from cloud_detect.providers import OCIProvider
 
 
@@ -13,3 +15,39 @@ def test_reading_invalid_vendor_file():
     assert provider.check_vendor_file() is False
     provider.vendor_file = ''
     assert provider.check_vendor_file() is False
+
+
+@pytest.mark.asyncio
+async def test_valid_metadata_server_check(aresponses):
+    mock_host = 'testing_metadata_url.com'
+    aresponses.add(
+        mock_host, '/', 'GET',
+        response={'id': 'ocid1.instance.oc1.phx.abc', 'region': 'phx'},
+    )
+    aresponses.add(
+        mock_host, '/', 'GET',
+        response={'id': 'ocid1.instance.oc1.phx.abc', 'region': 'phx'},
+    )
+
+    provider = OCIProvider()
+    provider.metadata_url = f'https://{mock_host}'
+    provider.metadata_url_v2 = f'https://{mock_host}'
+    assert await provider.check_metadata_server() is True
+
+
+@pytest.mark.asyncio
+async def test_invalid_metadata_server_check(aresponses):
+    mock_host = 'testing_metadata_url.com'
+    aresponses.add(
+        mock_host, '/', 'GET',
+        response={'id': 'i-notoracle', 'region': 'somewhere'},
+    )
+    aresponses.add(
+        mock_host, '/', 'GET',
+        response={'id': 'i-notoracle', 'region': 'somewhere'},
+    )
+
+    provider = OCIProvider()
+    provider.metadata_url = f'https://{mock_host}'
+    provider.metadata_url_v2 = f'https://{mock_host}'
+    assert await provider.check_metadata_server() is False