Add a country's continent to alert email template data

.devcontainer/docker-compose.yml

@@ -23,6 +23,7 @@ services:
       - "5354:5354/udp"
       - "51820:51820/udp"
       - "8082:8082"
+      - "8083:8083"
     # Overrides default command so things don't shut down after the process ends.
     entrypoint: /usr/local/share/docker-init.sh
     command: sleep infinity

canarytokens/channel_output_email.py

@@ -323,6 +323,11 @@ def format_report_html(
                 BasicDetails[field_name] = BasicDetails["src_data"].pop(field_name)
         BasicDetails.pop("src_data")
 
+        if "useragent" in BasicDetails and not BasicDetails["useragent"]:
+            BasicDetails.pop("useragent")
+        if "src_ip" in BasicDetails and not BasicDetails["src_ip"]:
+            BasicDetails.pop("src_ip")
+
         rendered_html = Template(template_path.open().read()).render(
             Title=EmailOutputChannel.DESCRIPTION,
             Intro=EmailOutputChannel.format_report_intro(details),

canarytokens/models.py

@@ -49,7 +49,7 @@
     CANARY_IMAGE_URL,
     MEMO_MAX_CHARACTERS,
 )
-from canarytokens.utils import prettify_snake_case, dict_to_csv
+from canarytokens.utils import prettify_snake_case, dict_to_csv, get_src_ip_continent
 
 CANARYTOKEN_RE = re.compile(
     ".*([" + "".join(CANARYTOKEN_ALPHABET) + "]{" + str(CANARYTOKEN_LENGTH) + "}).*",
@@ -1308,7 +1308,7 @@ def get_additional_data_for_notification(self) -> Dict[str, Any]:
             Dict[str, str]: Key value pairs to include in webhook info.
         """
 
-        additional_info = json_safe_dict(
+        additional_data = json_safe_dict(
             self,
             exclude=(
                 "time_of_hit",
@@ -1318,14 +1318,24 @@ def get_additional_data_for_notification(self) -> Dict[str, Any]:
                 "token_type",
             ),
         )
-        if "additional_info" in additional_info:
-            additional_info.update(**additional_info.pop("additional_info"))
+        if "additional_data" in additional_data:
+            additional_data.update(**additional_data.pop("additional_data"))
         for key, replacement in [("l", "location"), ("r", "referer")]:
-            if key in additional_info:
-                additional_info[replacement] = additional_info.pop(key)
+            if key in additional_data:
+                additional_data[replacement] = additional_data.pop(key)
             if self.src_data and key in self.src_data:
                 self.src_data[replacement] = self.src_data[key]
-        return additional_info
+
+        if additional_data.get("geo_info", {}).get("country") is not None:
+            additional_data["geo_info"]["continent"] = get_src_ip_continent(
+                self.geo_info.country
+            )
+
+        time = datetime.utcnow()
+        additional_data["time_hm"] = time.strftime("%H:%M")
+        additional_data["time_ymd"] = time.strftime("%y/%m/%d")
+
+        return additional_data
 
 
 class AzureIDTokenHit(TokenHit):

canarytokens/utils.py

@@ -2,6 +2,8 @@
 from pathlib import Path
 from typing import Any, Literal, Union
 
+import pycountry_convert
+
 
 def dict_to_csv(d: dict) -> str:
     """Convert dict to CSV"""
@@ -74,3 +76,22 @@ def get_deployed_commit_sha(commit_sha_file: Path = Path("/COMMIT_SHA")):
 #         return wrapper
 
 #     return inner
+
+
+def get_src_ip_continent(country: str) -> str:
+    """Helper function that returns the continent of country given it's ISO 3166-2 code.
+
+    Args:
+        country (str): ISO 3166-2 code
+
+    Returns:
+        str: A two character code representing a continent
+    """
+    # AQ is the ISO 3166-2 code for Antarctica, and is returned from IPinfo,
+    # but it's not included in pycountry_convert.
+    if country == "AQ":
+        return "AN"
+    try:
+        return pycountry_convert.country_alpha2_to_continent_code(country)
+    except KeyError:
+        return "NO_CONTINENT"

pyproject.toml

@@ -33,6 +33,7 @@ azure-core = "^1.23.1"
 azure-identity = "^1.10.0"
 cssutils = "^1.0.2"
 advocate = {git = "https://github.com/JordanMilne/Advocate.git"}
+pycountry-convert = "^0.7.2"
 
 [tool.poetry.extras]
 web = ["fastapi", "uvicorn", "sentry-sdk"]

tests/units/test_models.py

@@ -20,6 +20,7 @@
     DownloadContentTypes,
     DownloadMSWordResponse,
     GeoIPBogonInfo,
+    GeoIPInfo,
     LegacyTokenHistory,
     LegacyTokenHit,
     Log4ShellTokenHistory,
@@ -532,13 +533,42 @@ def test_get_additional_data_for_webhook(
                 src_ip="127.0.0.1",
                 is_tor_relay=True,
                 input_channel="HTTP",
-                **seed_data,
             )
         ]
     )
     assert expected_data == hist.latest_hit().get_additional_data_for_notification()
 
 
+@pytest.mark.parametrize(
+    "history_type, hit_type, seed_data",
+    [
+        (
+            WebBugTokenHistory,
+            WebBugTokenHit,
+            {
+                "geo_info": GeoIPInfo(
+                    ip="1.1.1.1", country="ZA", loc="-33.9778,18.6167"
+                ),
+            },
+        ),
+    ],
+)
+def test_get_additional_data_for_email(history_type, hit_type, seed_data):
+    hist = history_type(
+        hits=[
+            hit_type(
+                time_of_hit=20,
+                input_channel="HTTP",
+                **seed_data,
+            )
+        ]
+    )
+    additional_data = hist.latest_hit().get_additional_data_for_notification()
+    assert additional_data["geo_info"]["continent"] == "AF"
+    assert "time_hm" in additional_data
+    assert "time_ymd" in additional_data
+
+
 @pytest.mark.parametrize(
     "seed_data, expected_data",
     [

tests/units/test_utils.py

@@ -1,6 +1,7 @@
 from canarytokens.utils import (
     coerce_to_float,
     get_deployed_commit_sha,
+    get_src_ip_continent,
 )
 
 
@@ -19,3 +20,14 @@ def test_coerce_to_float():
     assert 10.3 == coerce_to_float("10.3")
     assert 10 == coerce_to_float("10")
     assert not coerce_to_float("notafloat")
+
+
+def test_get_src_ip_continent():
+    assert "AF" == get_src_ip_continent("ZA")
+    assert "AN" == get_src_ip_continent("AQ")
+    assert "AS" == get_src_ip_continent("CN")
+    assert "EU" == get_src_ip_continent("GB")
+    assert "NA" == get_src_ip_continent("US")
+    assert "OC" == get_src_ip_continent("AU")
+    assert "SA" == get_src_ip_continent("AR")
+    assert "NO_CONTINENT" == get_src_ip_continent("1234")