Port Azure ID token to v3

.github/workflows/test.yml

@@ -4,7 +4,7 @@ on:
   push:
     branches:
       - "T4627_py3_main"
-      - "T5281_port_cc_token"
+      - "T5285_port_azure_token"
 
 jobs:
   tests:
@@ -111,7 +111,7 @@ jobs:
           export CANARY_MAILGUN_API_KEY=${{ secrets.TESTING_MAILGUN_API_KEY }}
           export CANARY_SENTRY_ENVIRONMENT=ci
           cd tests
-          poetry run coverage run  --source=../canarytokens --omit="integration/test_custom_binary.py,integration/test_sql_server_token.py" -m pytest units --runv3
+          poetry run coverage run  --source=../canarytokens --omit="integration/test_custom_binary.py,integration/test_sql_server_token.py" -m pytest units --runv3 -v
 
       - name: Check coverage is over threshold percentage for unit tests
         # Here we check coverage info on all tests
@@ -128,13 +128,13 @@ jobs:
           sleep 10
           export TEST_NETWORK=`docker network ls | grep git | python -c "from sys import stdin; print(stdin.read().split()[1])"`
           export TEST_HOST=`docker network inspect $TEST_NETWORK | jq '.[0].IPAM.Config[0].Gateway' | sed 's/"//g'`
-          LIVE=False poetry run coverage run  --source=../canarytokens --omit=integration/test_custom_binary.py -m pytest integration --runv3
+          LIVE=False poetry run coverage run  --source=../canarytokens --omit=integration/test_custom_binary.py -m pytest integration --runv3 -v
 
       - name: Run integration tests (against V2)
         # Here we gather coverage info on integration tests.
         run: |
           cd tests
-          LIVE=True poetry run coverage run  --source=integration --omit=integration/test_custom_binary.py -m pytest integration --runv2
+          LIVE=True poetry run coverage run  --source=integration --omit=integration/test_custom_binary.py -m pytest integration --runv2 -v
 
       - name: Check coverage is over threshold percentage for integration tests
         # Here we check coverage info that all integration tests where indeed run.

canarytokens/authenticode.py

@@ -4,7 +4,9 @@
 from canarytokens.sign_file import authenticode_sign_binary
 
 
-def make_canary_authenticode_binary(nxdomain_token_url: str, filebody: bytes) -> bytes:
+def make_canary_authenticode_binary(
+    nxdomain_token_url: str, filebody: bytes
+) -> bytes:  # pragma: no cover
     """Takes in a nxdomain url (eg: http://{token}.nxdomain.tools) and bytes string (some binary to sign)
     and returns bytes (the signed binary).
 

canarytokens/awskeys.py

@@ -1,19 +1,12 @@
 import logging
 import re
-from typing import Literal, Optional, TypedDict
+from typing import Optional
 
 import requests
 from pydantic import HttpUrl
 
 from canarytokens import tokens
-
-
-class AWSKey(TypedDict):
-    access_key_id: str
-    secret_access_key: str
-    # TODO: make enum
-    region: str
-    output: Literal["json", "yaml", "yaml-stream", "text", "table"]
+from canarytokens.models import AWSKey
 
 
 def validate_record(server: str, token: tokens.Canarytoken) -> bool:

canarytokens/azurekeys.py

@@ -0,0 +1,50 @@
+import logging
+from typing import Optional
+
+import requests
+from pydantic import HttpUrl
+
+from canarytokens.models import AzureID
+from canarytokens import tokens
+
+
+def get_azure_id(
+    token: tokens.Canarytoken,
+    server: str,
+    cert_file_name: str,
+    azure_url: Optional[HttpUrl] = None,
+    app_id: Optional[str] = None,
+    cert: Optional[str] = None,
+    tenant_id: Optional[str] = None,
+    cert_name: Optional[str] = None,
+) -> AzureID:  # pragma: no cover
+    if app_id and cert and tenant_id and cert_name:
+        return AzureID(
+            **{
+                "app_id": app_id,
+                "cert": cert,
+                "tenant_id": tenant_id,
+                "cert_name": cert_name,
+                "cert_file_name": cert_file_name,
+            }
+        )
+    if not (token and server) or len(server) == 0:
+        logging.error("Empty values passed through to get_azure_id function.")
+        raise ValueError("get_azure_id requires token and server to be set.")
+    if not azure_url:
+        raise ValueError("get_azure_id requires azure_url to request from.")
+
+    data = {"token": token.value(), "domain": server}
+
+    resp = requests.post(url=azure_url, json=data)
+    resp.raise_for_status()
+    resp_json = resp.json()
+    return AzureID(
+        **{
+            "app_id": resp_json["app_id"],
+            "cert": resp_json["cert"],
+            "tenant_id": resp_json["tenant_id"],
+            "cert_name": resp_json["cert_name"],
+            "cert_file_name": cert_file_name,
+        }
+    )

canarytokens/canarydrop.py

@@ -112,6 +112,14 @@ class Canarydrop(BaseModel):
     aws_secret_access_key: Optional[str]
     aws_output: Optional[str] = Field(alias="output")
     aws_region: Optional[str] = Field(alias="region")
+
+    # Azure key specific stuff
+    app_id: Optional[str]
+    tenant_id: Optional[str]
+    cert: Optional[str]
+    cert_name: Optional[str]
+    cert_file_name: Optional[str]
+
     # HTTP style token specific stuff
     browser_scanner_enabled: Optional[bool]
     # Wireguard specific stuff

canarytokens/channel_http.py

@@ -127,6 +127,11 @@ def render_POST(self, request: Request):
             canarydrop.add_canarydrop_hit(token_hit=token_hit)
             self.dispatch(canarydrop=canarydrop, token_hit=token_hit)
             return b"success"
+        elif canarydrop.type == TokenTypes.AZURE_ID:
+            token_hit = Canarytoken._parse_azure_id_trigger(request)
+            canarydrop.add_canarydrop_hit(token_hit=token_hit)
+            self.dispatch(canarydrop=canarydrop, token_hit=token_hit)
+            return b"success"
         elif canarydrop.type in [
             TokenTypes.SLOW_REDIRECT,
             TokenTypes.WEB_IMAGE,

canarytokens/extendtoken.py

@@ -30,7 +30,7 @@ def __init__(
         password,
         card_name,
         token=None,
-    ):
+    ):  # pragma: no cover
         self.email = email
         self.token = token
         self.kind = "AMEX"
@@ -45,7 +45,9 @@ def __init__(
         self.token = req.json().get("token")
         self.refresh_token = req.json().get("refresh_token")
 
-    def _post_api(self, endpoint: str, data: Optional[str] = None) -> requests.Response:
+    def _post_api(
+        self, endpoint: str, data: Optional[str] = None
+    ) -> requests.Response:  # pragma: no cover
         """Performs a POST against the passed endpoint with the data passed"""
         headers = {
             "Content-Type": "application/json",
@@ -163,7 +165,7 @@ def get_virtual_cards(self) -> list[tuple[str, str]]:  # pragma: no cover
             )
         return cards
 
-    def get_card_info(self, card_id) -> Optional[dict[str, str]]:
+    def get_card_info(self, card_id) -> Optional[dict[str, str]]:  # pragma: no cover
         """Returns all the data about a passed card_id available"""
         req = self._get_api("https://v.paywithextend.com/virtualcards/" + card_id)
         return req.json()
@@ -228,7 +230,7 @@ def make_card(
         address: str,
         billing_zip: str,
         limit_cents: int = 100,
-    ) -> CreditCard:
+    ) -> CreditCard:  # pragma: no cover
         """Creates a new CreditCard via Extend's CreateVirtualCard API"""
         cc = self.get_parent_card_id()
         now_ts = datetime.datetime.now() - datetime.timedelta(days=1)
@@ -283,7 +285,7 @@ def create_credit_card(
         last_name: Optional[str] = None,
         address: Optional[str] = None,
         billing_zip: Optional[str] = None,
-    ) -> CreditCard:
+    ) -> CreditCard:  # pragma: no cover
         """Creates a cardholder and associated virtual card for the passed person, if not passed, will generate fake data to use"""
         fake_person = generate_person()
         if first_name is None:
@@ -305,13 +307,13 @@ def create_credit_card(
 
         return cc
 
-    def get_credit_card(self, id: str) -> CreditCard:
+    def get_credit_card(self, id: str) -> CreditCard:  # pragma: no cover
         """Abstract method to get a virtual credit card"""
         pass
 
-    def get_transaction_events(  # noqa C901  # pragma: no cover
+    def get_transaction_events(  # noqa C901
         self, since: Optional[datetime.datetime] = None
-    ) -> list:
+    ) -> list:  # pragma: no cover
         """Returns a list of recent transactions for the org"""
         txns = []
         req = self._get_api("https://api.paywithextend.com/events")