Merge pull request #795 from cisco-open/apigw

Add API GW login. Refactor session related files.

README.md

@@ -38,6 +38,7 @@ with create_manager_session(url=url, username=username, password=password) as se
     devices = session.api.devices.get()
     print(devices)
 ```
+
 **ManagerSession** extends [requests.Session](https://requests.readthedocs.io/en/latest/user/advanced/#session-objects) so all functionality from [requests](https://requests.readthedocs.io/en/latest/) library is avaiable to user, it also implements python [contextmanager](https://docs.python.org/3.8/library/contextlib.html#contextlib.contextmanager) and automatically frees server resources on exit.
 
 <details>
@@ -47,13 +48,15 @@ It is possible to configure **ManagerSession** prior sending any request.
 
 ```python
 from catalystwan.session import ManagerSession
+from catalystwan.vmanage_auth import vManageAuth
 
 url = "example.com"
 username = "admin"
 password = "password123"
 
 # configure session using constructor - nothing will be sent to target server yet
-session = ManagerSession(url=url, username=username, password=password)
+auth = vManageAuth(username, password)
+session = ManagerSession(url=url, auth=auth)
 # login and send requests
 session.login()
 session.get("/dataservice/device")
@@ -102,6 +105,25 @@ with create_manager_session(url=url, username=username, password=password, subdo
 
 </details>
 
+<details>
+    <summary> <b>Login using Api Gateway</b> <i>(click to expand)</i></summary>
+
+```python
+from catalystwan.session import create_apigw_session
+
+with create_apigw_session(
+    url="example.com",
+    client_id="client_id",
+    client_secret="client_secret",
+    org_name="Org-Name",
+    username="user",
+    mode="user",
+    token_duration=10,
+) as session:
+    devices = session.api.devices.get()
+    print(devices)
+```
+</details>
 
 
 ## API usage examples

catalystwan/api/administration.py

@@ -100,16 +100,16 @@ def update(self, user_update_request: UserUpdateRequest):
         """
         self._endpoints.update_user(user_update_request.username, user_update_request)
 
-    def update_password(self, username: str, new_password: str):
+    def update_password(self, username: str, new_password: str, current_user_password: str):
         """Updates exisiting user password
 
         Args:
             username (str): Name of the user
             new_password (str): New password for given user
         """
         update_password_request = UserUpdateRequest(
-            username=username, password=new_password, current_user_password=self.session.password
-        )  # type: ignore
+            username=username, password=new_password, current_user_password=current_user_password
+        )
         self._endpoints.update_password(username, update_password_request)
 
     def reset(self, username: str):

catalystwan/api/tenant_management_api.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, List
 
 from catalystwan.api.task_status_api import Task
 from catalystwan.endpoints.tenant_management import (
@@ -62,7 +62,7 @@ def update(self, tenant_update_request: TenantUpdateRequest) -> Tenant:
             tenant_id=tenant_update_request.tenant_id, tenant_update_request=tenant_update_request
         )
 
-    def delete(self, tenant_id_list: List[str], password: Optional[str] = None) -> Task:
+    def delete(self, tenant_id_list: List[str], password: str) -> Task:
         """Deletes tenants on vManage
 
         Args:
@@ -72,8 +72,6 @@ def delete(self, tenant_id_list: List[str], password: Optional[str] = None) -> T
         Returns:
             Task: Object representing tenant deletion process
         """
-        if password is None:
-            password = self.session.password
         delete_request = TenantBulkDeleteRequest(tenant_id_list=tenant_id_list, password=password)
         task_id = self._endpoints.delete_tenant_async_bulk(delete_request).id
         return Task(self.session, task_id)

catalystwan/apigw_auth.py

@@ -0,0 +1,93 @@
+import logging
+from typing import TYPE_CHECKING, Literal, Optional
+from urllib.parse import urlparse
+
+from pydantic import BaseModel, Field, PositiveInt
+from requests import HTTPError, PreparedRequest, post
+from requests.auth import AuthBase
+from requests.exceptions import JSONDecodeError
+
+from catalystwan.exceptions import CatalystwanException
+from catalystwan.response import ManagerResponse
+
+if TYPE_CHECKING:
+    from catalystwan.session import ManagerSession
+
+LoginMode = Literal["machine", "user", "session"]
+
+
+class ApiGwLogin(BaseModel):
+    client_id: str
+    client_secret: str
+    org_name: str
+    mode: Optional[LoginMode] = None
+    username: Optional[str] = None
+    session: Optional[str] = None
+    tenant_user: Optional[bool] = None
+    token_duration: PositiveInt = Field(default=10, description="in minutes")
+
+
+class ApiGwAuth(AuthBase):
+    """Attaches ApiGateway Authentication to the given Requests object.
+
+    1. Get a bearer token by sending a POST request to the /apigw/login endpoint.
+    2. Use the token in the Authorization header for subsequent requests.
+    """
+
+    def __init__(self, login: ApiGwLogin, logger: Optional[logging.Logger] = None):
+        self.login = login
+        self.token = ""
+        self.logger = logger or logging.getLogger(__name__)
+
+    def __call__(self, request: PreparedRequest) -> PreparedRequest:
+        self.handle_auth(request)
+        self.build_digest_header(request)
+        return request
+
+    def handle_auth(self, request: PreparedRequest) -> None:
+        if self.token == "":
+            self.authenticate(request)
+
+    def authenticate(self, request: PreparedRequest):
+        assert request.url is not None
+        url = urlparse(request.url)
+        base_url = f"{url.scheme}://{url.netloc}"
+        self.token = self.get_token(base_url, self.login)
+
+    def build_digest_header(self, request: PreparedRequest) -> None:
+        header = {
+            "sdwan-org": self.login.org_name,
+            "Authorization": f"Bearer {self.token}",
+        }
+        request.headers.update(header)
+
+    @staticmethod
+    def get_token(base_url: str, apigw_login: ApiGwLogin) -> str:
+        try:
+            response = post(
+                url=f"{base_url}/apigw/login",
+                verify=False,
+                json=apigw_login.model_dump(exclude_none=True),
+                timeout=10,
+            )
+            response.raise_for_status()
+            token = response.json()["token"]
+        except JSONDecodeError:
+            raise CatalystwanException(f"Incorrect response type from ApiGateway login request, ({response.text})")
+        except HTTPError as ex:
+            raise CatalystwanException(f"Problem with connection to ApiGateway login endpoint, ({ex})")
+        except KeyError as ex:
+            raise CatalystwanException(f"Not found token in login response from ApiGateway, ({ex})")
+        else:
+            if not token or not isinstance(token, str):
+                raise CatalystwanException("Failed to get bearer token")
+        return token
+
+    def __str__(self) -> str:
+        return f"ApiGatewayAuth(mode={self.login.mode})"
+
+    def logout(self, session: "ManagerSession") -> Optional[ManagerResponse]:
+        return None
+
+    def clear_tokens_and_cookies(self) -> None:
+        self.token = ""

catalystwan/endpoints/api_gateway.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel
+
+from catalystwan.endpoints import APIEndpoints, post
+
+
+class OnBoardClient(BaseModel):
+    client_id: str
+    client_secret: str
+    client_name: str
+
+
+class ApiGateway(APIEndpoints):
+    @post("/apigw/config/reload")
+    def configuration_reload(self) -> None:
+        """After launching the API Gateway, SSP can use the API
+        and bearer authentication header with provision access token obtained
+        in the step above. It reloads the configuration from S3 bucket, Secrets Manager
+        and reset the RDS connection pool."""
+        ...
+
+    @post("/apigw/client/registration")
+    def on_board_client(self, payload: OnBoardClient) -> None:
+        ...

catalystwan/response.py

@@ -139,6 +139,7 @@ class ManagerResponse(Response, APIEndpointClientResponse):
     def __init__(self, response: Response):
         self.__dict__.update(response.__dict__)
         self.jsessionid_expired = self._detect_expired_jsessionid()
+        self.api_gw_unauthorized = self._detect_apigw_unauthorized()
         try:
             self.payload = JsonPayload(response.json())
         except JSONDecodeError:
@@ -163,6 +164,10 @@ def _parse_set_cookie_from_headers(self) -> RequestsCookieJar:
         jar.update(parse_cookies_to_dict(cookies_string))
         return jar
 
+    def _detect_apigw_unauthorized(self) -> bool:
+        """Determines if server sent unauthorized response"""
+        return self.status_code == 401 and self.json().get("message", "") == "failed to validate user"
+
     def info(self, history: bool = False) -> str:
         """Returns human readable string containing Request-Response contents
         Args: