Merge branch 'develop' into issue-43/remove-auto-appended-campaign-id

README.md

@@ -183,6 +183,22 @@ campaign.
 | first_report | First report (click) generated by a targeted user. Format: "YYYY-MM-DDThh:mm.ss" | datetime |
 | total_num_reports | Total number of user reports received during a campaign. | integer |
 
+## Campaign Summary Field Dictionary ##
+
+The following JSON data is exported by `gophish-export`. Campaign summaries for
+all assessments in a campaign is reported in the following format.
+
+| Name | Description | Type |
+|------|-------------|:----:|
+| subject | The subject line for a Gophish-generated email. | string |
+| sender | The from address of a Gophish-generated email. | string |
+| start_date | The start date of a campaign. | datetime |
+| end_date | The end date of a campaign. | datetime |
+| redirect | The URL the Gophish-generated email will redirect to. | string |
+| clicks | The total number of clicks reported by a campaign. | integer |
+| unique_clicks | The total number of clicks generated by unique users. | integer |
+| percent_clicks | The percentage of emails sent versus how many were clicks by a targeted user | float |
+
 ## Contributing ##
 
 We welcome contributions!  Please see [`CONTRIBUTING.md`](CONTRIBUTING.md) for

src/_version.py

@@ -1,2 +1,2 @@
 """This file defines the version of this project."""
-__version__ = "0.0.5"
+__version__ = "0.0.6"

src/models/models.py

@@ -250,3 +250,32 @@ def parse(cls, json):
             elif key in cls._valid_properties:
                 setattr(campaign, key, val)
         return campaign
+
+
+class Click(Model):
+    """The Click class represents a Click object generated by the Gophish API when a user clicks on an email."""
+
+    _valid_properties = {
+        "message": None,
+        "user": None,
+        "source_ip": None,
+        "time": None,
+        "application": None,
+    }
+
+    def __init__(self, **kwargs):
+        """Create a new click instance."""
+        for key, default in Click._valid_properties.items():
+            setattr(self, key, kwargs.get(key, default))
+
+    @classmethod
+    def parse(cls, json):
+        """Parse click json."""
+        click = cls()
+        for key, val in json.items():
+            if key in cls._valid_properties:
+                setattr(click, key, val)
+
+    def __getitem__(self, item):
+        """Get item by attribute name."""
+        return getattr(self, item)

src/tools/gophish_export.py

@@ -21,6 +21,7 @@
 import hashlib
 import json
 import logging
+import re
 import sys
 from typing import Dict
 
@@ -251,6 +252,81 @@ def get_application(rawEvent):
     return application
 
 
+def find_unique_target_clicks_count(clicks):
+    """Return the number of unique clicks in a click set."""
+    uniq_users = set()
+    for click in clicks:
+        uniq_users.add(click["user"])
+    return len(uniq_users)
+
+
+def write_campaign_summary(api, assessment_id):
+    """Output a campaign summary report to JSON, console, and a text file."""
+    campaign_ids = get_campaign_ids(api, assessment_id)
+    campaign_data_template = "campaign_data.json"
+    campaign_summary_json = f"{assessment_id}_campaign_data.json"
+    campaign_summary_textfile = f"{assessment_id}_summary_{datetime.strftime(datetime.now(), '%Y-%m-%dT%H:%M:%S')}.txt"
+
+    with open(campaign_data_template) as template:
+        campaign_data = json.load(template)
+
+    logging.info("Writing campaign summary report to %s", campaign_summary_textfile)
+    file_out = open(campaign_summary_textfile, "w+")
+    file_out.write("Campaigns for Assessment: " + assessment_id)
+
+    regex = re.compile(r"^.*_(?P<level>level-[1-6])$")
+    for campaign_id in campaign_ids:
+        campaign = api.campaigns.get(campaign_id)
+        match = regex.fullmatch(campaign.name)
+        if match:
+            level = match.group("level")
+        else:
+            logging.warn(
+                "Encountered campaign (%s) that is unable to be processed for campaign summary export. \n"
+                "Campaign name is not properly suffixed with the campaign level number (e.g. '_level-1')\n"
+                "Skipping campaign",
+                campaign.name,
+            )
+            continue
+
+        logging.info(level)
+        clicks = get_click_data(api, campaign_id)
+
+        total_clicks = api.campaigns.summary(campaign_id=campaign_id).stats.clicked
+        unique_clicks = find_unique_target_clicks_count(clicks)
+        if total_clicks > 0:
+            percent_clicks = unique_clicks / float(total_clicks)
+        else:
+            percent_clicks = 0.0
+        campaign_data[level]["subject"] = campaign.template.subject
+        campaign_data[level]["sender"] = campaign.smtp.from_address
+        campaign_data[level]["start_date"] = campaign.launch_date
+        campaign_data[level]["end_date"] = campaign.completed_date
+        campaign_data[level]["redirect"] = campaign.url
+        campaign_data[level]["clicks"] = total_clicks
+        campaign_data[level]["unique_clicks"] = unique_clicks
+        campaign_data[level]["percent_clicks"] = percent_clicks
+
+        file_out.write("\n")
+        file_out.write("-" * 50)
+        file_out.write("\nCampaign: %s" % campaign.name)
+        file_out.write("\nSubject: %s" % campaign_data[level]["subject"])
+        file_out.write("\nSender: %s" % campaign_data[level]["sender"])
+        file_out.write("\nStart Date: %s" % campaign_data[level]["start_date"])
+        file_out.write("\nEnd Date: %s" % campaign_data[level]["end_date"])
+        file_out.write("\nRedirect: %s" % campaign_data[level]["redirect"])
+        file_out.write("\nClicks: %i" % campaign_data[level]["clicks"])
+        file_out.write("\nUnique Clicks: %i" % campaign_data[level]["unique_clicks"])
+        file_out.write(
+            "\nPercentage Clicks: %f" % campaign_data[level]["percent_clicks"]
+        )
+
+    file_out.close()
+    logging.info("Writing out summary JSON to %s", campaign_summary_json)
+    with open(campaign_summary_json, "w") as fp:
+        json.dump(campaign_data, fp, indent=4)
+
+
 def export_user_reports(api, assessment_id):
     """Build and export a user_report JSON file for each campaign in an assessment."""
     campaign_ids = get_campaign_ids(api, assessment_id)
@@ -275,9 +351,13 @@ def export_user_reports(api, assessment_id):
         # get_campaign_ids() returns integers, but user_report_doc["campaign"]
         # expects a string
         user_report_doc["campaign"] = str(campaign_id)
-        user_report_doc["first_report"] = datetime.strftime(
-            first_report, "%Y-%m-%dT%H:%M:%S"
-        )
+        if first_report is not None:
+            user_report_doc["first_report"] = datetime.strftime(
+                first_report, "%Y-%m-%dT%H:%M:%S"
+            )
+        else:
+            user_report_doc["first_report"] = "No clicks reported"
+
         user_report_doc["total_num_reports"] = api.campaigns.summary(
             campaign_id=campaign_id
         ).stats.clicked
@@ -330,7 +410,7 @@ def main() -> None:
         logging.info(f'Data written to data_{args["ASSESSMENT_ID"]}.json')
 
         export_user_reports(api, args["ASSESSMENT_ID"])
-
+        write_campaign_summary(api, args["ASSESSMENT_ID"])
     else:
         logging.error(
             f'Assessment "{args["ASSESSMENT_ID"]}" does not exist in GoPhish.'

tests/conftest.py

@@ -13,7 +13,16 @@
 import pytest
 
 # cisagov Libraries
-from models.models import SMTP, Assessment, Campaign, Group, Page, Target, Template
+from models.models import (
+    SMTP,
+    Assessment,
+    Campaign,
+    Click,
+    Group,
+    Page,
+    Target,
+    Template,
+)
 
 """Support items for test_modules.py """
 
@@ -298,6 +307,50 @@ def email_target_json():
     return targets
 
 
+@pytest.fixture
+def multiple_click_object():
+    """Return a list of clicks to match the correct number of unique users."""
+    clicks = list()
+    for x in range(0, 2):
+        clicks.append(
+            Click(
+                message="Testing",
+                user="jane.smith1@domain.tld",
+                source_ip="10.0.0.0",
+                time="01/01/2025 13:00",
+                application="NA",
+            )
+        )
+        clicks.append(
+            Click(
+                message="Testing",
+                user="john.doe1@domain.tld",
+                source_ip="10.0.0.1",
+                time="01/01/2025 13:00",
+                application="NA",
+            )
+        )
+        clicks.append(
+            Click(
+                message="Testing",
+                user="jane.smith2@domain.tld",
+                source_ip="10.0.0.2",
+                time="01/01/2025 13:00",
+                application="NA",
+            )
+        )
+        clicks.append(
+            Click(
+                message="Testing",
+                user="john.doe2@domain.tld",
+                source_ip="10.0.0.3",
+                time="01/01/2025 13:00",
+                application="NA",
+            )
+        )
+    return clicks
+
+
 def pytest_addoption(parser):
     """Add new commandline options to pytest."""
     parser.addoption(

tests/test_tools.py

@@ -7,7 +7,11 @@
 
 # cisagov Libraries
 from tools.gophish_complete import get_campaign_id
-from tools.gophish_export import assessment_exists, export_targets
+from tools.gophish_export import (
+    assessment_exists,
+    export_targets,
+    find_unique_target_clicks_count,
+)
 
 
 class TestComplete:
@@ -43,9 +47,13 @@ def test_assessment_exists_found(self, mock_api, multiple_campaign_object):
     def test_assessment_exists_not_found(self, mock_api, multiple_campaign_object):
         """Verify False is returned when assessment is not in GoPhish."""
         mock_api.campaigns.get.return_value = multiple_campaign_object
-
         assert assessment_exists(mock_api, "RVXXX3") is False
 
+    @patch("tools.connect")
+    def test_find_unique_target_clicks_count(self, mock_api, multiple_click_object):
+        """Verify that the correct number of unique users in a click list is found."""
+        assert find_unique_target_clicks_count(multiple_click_object) == 4
+
     def mock_get_group_ids(self, s, group_object):
         """Return a mock list of GoPhish group objects."""
         return group_object