This commit is contained in:
49
sk/azure.py
49
sk/azure.py
@@ -12,8 +12,53 @@ from cryptography import x509
|
|||||||
import hashlib
|
import hashlib
|
||||||
import base64
|
import base64
|
||||||
|
|
||||||
|
DEVOPS_SCOPE = "https://app.vssps.visualstudio.com/.default"
|
||||||
|
|
||||||
|
def get_token(
|
||||||
|
tenant_id: str | None = None,
|
||||||
|
client_id: str | None = None,
|
||||||
|
client_secret: str | None = None,
|
||||||
|
pem_path: str | None = None
|
||||||
|
) -> str:
|
||||||
|
"""
|
||||||
|
Obtain a token for DevOps using DefaultAzureCredential.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
if tenant_id and client_id and client_secret:
|
||||||
|
from azure.identity import ClientSecretCredential
|
||||||
|
return ClientSecretCredential(
|
||||||
|
tenant_id=tenant_id,
|
||||||
|
client_id=client_id,
|
||||||
|
client_secret=client_secret
|
||||||
|
).get_token(DEVOPS_SCOPE).token
|
||||||
|
elif tenant_id and client_id and pem_path:
|
||||||
|
from azure.identity import CertificateCredential
|
||||||
|
return CertificateCredential(
|
||||||
|
tenant_id=tenant_id,
|
||||||
|
client_id=client_id,
|
||||||
|
certificate_path=pem_path
|
||||||
|
).get_token(DEVOPS_SCOPE).token
|
||||||
|
else:
|
||||||
|
from azure.identity import DefaultAzureCredential
|
||||||
|
return DefaultAzureCredential().get_token(DEVOPS_SCOPE).token
|
||||||
|
except ImportError:
|
||||||
|
if tenant_id and client_id and client_secret:
|
||||||
|
return secret_credentials_auth(
|
||||||
|
tenant_id=tenant_id,
|
||||||
|
client_id=client_id,
|
||||||
|
client_secret=client_secret
|
||||||
|
)
|
||||||
|
elif tenant_id and client_id and pem_path:
|
||||||
|
return certificate_credentials_auth(
|
||||||
|
tenant_id=tenant_id,
|
||||||
|
client_id=client_id,
|
||||||
|
pem_path=pem_path
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
raise ValueError("Either client_secret or pem_path must be provided, if no azure-identity package is installed.")
|
||||||
|
|
||||||
def secret_credentials_auth(
|
def secret_credentials_auth(
|
||||||
scope: str = "https://app.vssps.visualstudio.com/.default",
|
scope: str = DEVOPS_SCOPE,
|
||||||
tenant_id: str = os.environ.get("AZURE_TENANT_ID", ""),
|
tenant_id: str = os.environ.get("AZURE_TENANT_ID", ""),
|
||||||
client_id: str = os.environ.get("AZURE_CLIENT_ID", ""),
|
client_id: str = os.environ.get("AZURE_CLIENT_ID", ""),
|
||||||
client_secret: str = os.environ.get("AZURE_CLIENT_SECRET")
|
client_secret: str = os.environ.get("AZURE_CLIENT_SECRET")
|
||||||
@@ -33,7 +78,7 @@ def secret_credentials_auth(
|
|||||||
return r.json().get("access_token", "")
|
return r.json().get("access_token", "")
|
||||||
|
|
||||||
def certificate_credentials_auth(
|
def certificate_credentials_auth(
|
||||||
scope: str = "https://app.vssps.visualstudio.com/.default",
|
scope: str = DEVOPS_SCOPE,
|
||||||
tenant_id: str = os.environ.get("AZURE_TENANT_ID", ""),
|
tenant_id: str = os.environ.get("AZURE_TENANT_ID", ""),
|
||||||
client_id: str = os.environ.get("AZURE_CLIENT_ID", ""),
|
client_id: str = os.environ.get("AZURE_CLIENT_ID", ""),
|
||||||
pem_path: str = os.environ.get("AZURE_CLIENT_CERTIFICATE_PATH", "")
|
pem_path: str = os.environ.get("AZURE_CLIENT_CERTIFICATE_PATH", "")
|
||||||
|
|||||||
@@ -108,8 +108,6 @@ class DevOps():
|
|||||||
|
|
||||||
class Organization(DevOps):
|
class Organization(DevOps):
|
||||||
def __init__(self, org_url: str, token: str | None = None, api_version: str = DEVOPS_API_VERSION):
|
def __init__(self, org_url: str, token: str | None = None, api_version: str = DEVOPS_API_VERSION):
|
||||||
if token is None:
|
|
||||||
token = DefaultAzureCredential().get_token(DEVOPS_SCOPE).token
|
|
||||||
super().__init__(org_url, token, api_version)
|
super().__init__(org_url, token, api_version)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
|
|||||||
6
tests.py
6
tests.py
@@ -1,12 +1,12 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
import unittest
|
import unittest
|
||||||
import requests
|
import requests
|
||||||
from azure.identity import DefaultAzureCredential
|
from sk.devops import Organization, Repository, Project, Item
|
||||||
from sk.devops import DEVOPS_SCOPE, Organization, Repository, Project, Item
|
from sk.azure import get_token
|
||||||
|
|
||||||
# Get the token outside the test class to speed up tests.
|
# Get the token outside the test class to speed up tests.
|
||||||
# Each Unit test instantinates the class, so doing it here avoids repeated authentication.
|
# Each Unit test instantinates the class, so doing it here avoids repeated authentication.
|
||||||
token = DefaultAzureCredential().get_token(DEVOPS_SCOPE).token
|
token = get_token()
|
||||||
|
|
||||||
class Test01(unittest.TestCase):
|
class Test01(unittest.TestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
|
|||||||
Reference in New Issue
Block a user