Skip to content
This repository has been archived by the owner on Feb 6, 2023. It is now read-only.

Commit

Permalink
feat: Add username/password auth
Browse files Browse the repository at this point in the history
update httpx
  • Loading branch information
jjeff07 committed May 25, 2022
1 parent 07c04dd commit 420b6b1
Show file tree
Hide file tree
Showing 5 changed files with 233 additions and 187 deletions.
25 changes: 17 additions & 8 deletions ipfabric/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from urllib.parse import urljoin

from httpx import Client
from ipfabric_httpx_auth import PasswordCredentials, HeaderApiKey
from pydantic import BaseSettings

from ipfabric import models
Expand All @@ -11,10 +12,12 @@


class Settings(BaseSettings):
ipf_url: str = None
ipf_token: str = None
ipf_url: str = ''
ipf_token: str = ''
ipf_verify: bool = True
ipf_dev: bool = False
ipf_username: str = ''
ipf_password: str = ''

class Config:
env_file = ".env"
Expand All @@ -32,7 +35,9 @@ def __init__(
self,
base_url: Optional[str] = None,
token: Optional[str] = None,
snapshot_id: str = DEFAULT_ID,
snapshot_id: Optional[str] = DEFAULT_ID,
username: Optional[str] = None,
password: Optional[str] = None,
**kwargs,
):
"""
Expand All @@ -43,20 +48,24 @@ def __init__(
:param kwargs: dict: Keyword args to pass to httpx
"""
with Settings() as settings:
base_url = base_url or settings.ipf_url
if settings.ipf_dev:
kwargs["base_url"] = urljoin(base_url or settings.ipf_url, "v1/")
kwargs["base_url"] = urljoin(base_url, "v1/")
else:
kwargs["base_url"] = urljoin(base_url or settings.ipf_url, "api/v1/")
kwargs["base_url"] = urljoin(base_url, "api/v1/")
kwargs["verify"] = kwargs.get("verify") if "verify" in kwargs else settings.ipf_verify
token = token or settings.ipf_token
username = username or settings.ipf_username
password = password or settings.ipf_password

if not kwargs["base_url"]:
raise RuntimeError("IP Fabric base_url not provided or IPF_URL not set")
if not token:
raise RuntimeError("IP Fabric token not provided or IPF_TOKEN not set")
if not token and not (username and password):
raise RuntimeError("IP Fabric Token or Username/Password not provided.")

super().__init__(**kwargs)
self.headers.update({"Content-Type": "application/json", "X-API-Token": token})
self.headers.update({"Content-Type": "application/json"})
self.auth = HeaderApiKey(token) if token else PasswordCredentials(base_url, username, password)

# Request IP Fabric for the OS Version, by doing that we are also ensuring the token is valid
self.os_version = self.fetch_os_version()
Expand Down
5 changes: 3 additions & 2 deletions ipfabric/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,8 @@ def wrapper(self, url, *args, **kwargs):

class IPFClient(IPFabricAPI):
def __init__(
self, base_url: Optional[str] = None, token: Optional[str] = None, snapshot_id: str = DEFAULT_ID, **kwargs
self, base_url: Optional[str] = None, token: Optional[str] = None, snapshot_id: str = DEFAULT_ID,
username: Optional[str] = None, password: Optional[str] = None, **kwargs
):
"""
Initializes the IP Fabric Client
Expand All @@ -36,7 +37,7 @@ def __init__(
:param snapshot_id: str: IP Fabric snapshot ID to use by default for database actions - defaults to '$last'
:param kwargs: dict: Keyword args to pass to httpx
"""
super().__init__(base_url, token, snapshot_id, **kwargs)
super().__init__(base_url, token, snapshot_id, username, password, **kwargs)
self.inventory = models.Inventory(client=self)
self.intent = Intent(client=self)

Expand Down
Loading

0 comments on commit 420b6b1

Please sign in to comment.