1218583375

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #40 from cisagov/add_campaign_summary_export

Run Details

93 of 380 branches covered (24.47%)

Branch coverage included in aggregate %.

15 of 71 new or added lines in 3 files covered. (21.13%)

1 existing line in 1 file now uncovered.

292 of 1259 relevant lines covered (23.19%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

23.41

/src/tools/gophish_export.py

"""Export all the data from an assessment within GoPhish into a single JSON file.

Usage:
  gophish-export [--log-level=LEVEL] ASSESSMENT_ID SERVER API_KEY
  gophish-export (-h | --help)
  gophish-export --version

Options:
  API_KEY                   GoPhish API key.
  ASSESSMENT_ID             ID of the assessment to export data from.
  SERVER                    Full URL to GoPhish server.
  -h --help                 Show this screen.
  --version                 Show version.
  -l --log-level=LEVEL      If specified, then the log level will be set to
                            the specified value.  Valid values are "debug", "info",
                            "warning", "error", and "critical". [default: info]
"""

# Standard Python Libraries
from datetime import datetime
import hashlib
import json
import logging
import re
import sys
from typing import Dict

# Third-Party Libraries
from docopt import docopt
import httpagentparser
import requests

# cisagov Libraries
from tools.connect import connect_api

from ._version import __version__

# Disable "Insecure Request" warning: GoPhish uses a self-signed certificate
# as default for https connections, which can not be  verified by a third
# party; thus, an SSL insecure request warning is produced.
requests.packages.urllib3.disable_warnings()


def assessment_exists(api, assessment_id):
    """Check if GoPhish has at least one campaign for designated assessment.

    Args:
        api (GoPhish API): Connection to GoPhish server via the API.
        assessment_id (string): Assessment identifier to get campaigns from.

    Returns:
        boolean: Indicates if a campaign is found starting with assessment_id.
    """
    allCampaigns = api.campaigns.get()
    for campaign in allCampaigns:
        if campaign.name.startswith(assessment_id):
            return True

    return False


def export_targets(api, assessment_id):
    """Add all targets to a list.

    Achieved by pulling the group IDs for any group starting with
    the assessment id. The targets within the group are then parsed
    into a targets list of target dicts. Each target dict includes a
    sha256 hash of the target's email and assessment id with any labels.

    Args:
        api (GoPhish API): Connection to GoPhish server via the API.
        assessment_id (string): Assessment identifier to get campaigns from.

    Returns:
        List of targets from the assessment's group(s).
    """
    groupIDs = get_group_ids(api, assessment_id)

    targets = list()

    for group_id in groupIDs:
        # Gets target list for parsing.
        raw_targets = api.groups.get(group_id).as_dict()["targets"]

        for raw_target in raw_targets:

            target = dict()

            target["id"] = hashlib.sha256(
                raw_target["email"].encode("utf-8")
            ).hexdigest()
            target["customer_defined_labels"] = dict()

            if "position" in raw_target:
                target["customer_defined_labels"][assessment_id] = [
                    raw_target["position"]
                ]

            targets.append(target)

    logging.info(f"{len(targets)} email targets found for assessment {assessment_id}.")

    return targets


def get_group_ids(api, assessment_id):
    """Return a list of group IDs for all groups starting with specified assessment_id."""
    rawGroup = api.groups.get()  # Holds raw list of campaigns from GoPhish.
    groups = list()  # Holds list of campaign IDs that match the assessment.

    for group in rawGroup:
        group = group.as_dict()
        if group["name"].startswith(assessment_id):
            groups.append(group["id"])

    return groups


def export_campaigns(api, assessment_id):
    """Add all the campaigns' data for an assessment to a list.

    Args:
        api (GoPhish API): Connection to GoPhish server via the API.
        assessment_id (string): Assessment identifier to get campaigns from.

    Returns:
        List of the assessment's campaigns with data.
    """
    campaignIDs = get_campaign_ids(api, assessment_id)
    campaigns = list()

    for campaign_id in campaignIDs:
        campaigns.append(get_campaign_data(api, campaign_id))

    logging.info(f"{len(campaigns)} campaigns found for assessment {assessment_id}.")

    return campaigns


def get_campaign_ids(api, assessment_id):
    """Return a list of campaign IDs for all campaigns starting with specified assessment_id."""
    rawCampaigns = api.campaigns.get()  # Holds raw list of campaigns from GoPhish.
    campaigns = list()  # Holds list of campaign IDs that match the assessment.

    for campaign in rawCampaigns:
        campaign = campaign.as_dict()
        if campaign["name"].startswith(assessment_id):
            campaigns.append(campaign["id"])

    return campaigns


def get_campaign_data(api, campaign_id):
    """Return campaign metadata for the given campaign ID."""
    campaign = dict()

    # Pulls the campaign data as dict from GoPhish.
    rawCampaign: dict = api.campaigns.get(campaign_id).as_dict()

    campaign["id"] = rawCampaign["name"]

    campaign["start_time"] = rawCampaign["launch_date"]
    campaign["end_time"] = rawCampaign["completed_date"]
    campaign["url"] = rawCampaign["url"]

    campaign["subject"] = rawCampaign["template"]["subject"]

    # Get the template ID from the GoPhish template name.
    campaign["template"] = (
        api.templates.get(rawCampaign["template"]["id"]).as_dict()["name"].split("-")[2]
    )

    campaign["clicks"] = get_click_data(api, campaign_id)

    # Get the e-mail send status from GoPhish.
    campaign["status"] = get_email_status(api, campaign_id)

    return campaign


def get_click_data(api, campaign_id):
    """Return a list of all clicks for a given campaign."""
    rawEvents = api.campaigns.get(campaign_id).as_dict()["timeline"]
    clicks = list()  # Holds list of all users that clicked.

    for rawEvent in rawEvents:
        if rawEvent["message"] == "Clicked Link":
            click = dict()

            # Builds out click document.
            click["user"] = hashlib.sha256(
                rawEvent["email"].encode("utf-8")
            ).hexdigest()
            click["source_ip"] = rawEvent["details"]["browser"]["address"]

            click["time"] = rawEvent["time"]

            click["application"] = get_application(rawEvent)

            clicks.append(click)

    return clicks


def get_email_status(api, campaign_id):
    """Return the email send status and time."""
    rawEvents = api.campaigns.get(campaign_id).as_dict()["timeline"]
    status = list()
    for rawEvent in rawEvents:
        email = dict()

        if rawEvent["message"] == "Email Sent":
            email["user"] = hashlib.sha256(
                rawEvent["email"].encode("utf-8")
            ).hexdigest()

            email["time"] = rawEvent["time"]

            email["status"] = "SUCCESS"

        elif rawEvent["message"] == "Error Sending Email":

            email["user"] = hashlib.sha256(
                rawEvent["email"].encode("utf-8")
            ).hexdigest()

            # Trim microseconds before converting to datetime.
            rawEvent["time"] = datetime.strptime(
                rawEvent["time"].split(".")[0], "%Y-%m-%dT%H:%M:%S"
            )
            email["time"] = rawEvent["time"]

            email["status"] = "Failed"

        if email:
            status.append(email)

    return status


def get_application(rawEvent):
    """Return application details."""
    application = dict()

    application["external_ip"] = rawEvent["details"]["browser"]["address"]

    # Process user agent string.
    userAgent = rawEvent["details"]["browser"]["user-agent"]
    application["name"] = httpagentparser.detect(userAgent)["platform"]["name"]
    application["version"] = httpagentparser.detect(userAgent)["platform"]["version"]

    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)

    for campaign_id in campaign_ids:
        first_report = None
        user_report_doc = dict()
        campaign = get_campaign_data(api, campaign_id)

        # iterate over clicks and find the earliest click
        for click in campaign["clicks"]:
            click_time = datetime.strptime(
                click["time"].split(".")[0], "%Y-%m-%dT%H:%M:%S"
            )
            if first_report is None or click_time < first_report:
                first_report = click_time

        # The "customer" field is a placeholder added for operator convenience when
        # working with the JSON file created.
        user_report_doc["customer"] = ""
        user_report_doc["assessment"] = assessment_id
        # get_campaign_ids() returns integers, but user_report_doc["campaign"]
        # expects a string
        user_report_doc["campaign"] = str(campaign_id)
        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

        logging.info(
            f"Writing out user report for campaign {campaign_id} in assessment {assessment_id}"
        )

        with open(f"{assessment_id}_{campaign_id}_user_report_doc.json", "w") as fp:
            json.dump(user_report_doc, fp, indent=4)


def main() -> None:
    """Set up logging, connect to API, export all assessment data."""
    args: Dict[str, str] = docopt(__doc__, version=__version__)

    # Set up logging
    log_level = args["--log-level"]
    try:
        logging.basicConfig(
            format="\n%(levelname)s: %(message)s", level=log_level.upper()
        )
    except ValueError:
        logging.critical(
            f'"{log_level}"is not a valid logging level. Possible values are debug, info, warning, and error.'
        )
        sys.exit(1)

    else:
        # Connect to API
        try:
            api = connect_api(args["API_KEY"], args["SERVER"])
            logging.debug(f'Connected to: {args["SERVER"]}')
        except Exception as e:
            logging.critical(e.args[0])
            sys.exit(1)

    if assessment_exists(api, args["ASSESSMENT_ID"]):
        assessment_dict: Dict = dict()

        # Add targets list to assessment dict.
        assessment_dict["targets"] = export_targets(api, args["ASSESSMENT_ID"])

        # Add campaigns list to the assessment dict.
        assessment_dict["campaigns"] = export_campaigns(api, args["ASSESSMENT_ID"])

        with open(f'data_{args["ASSESSMENT_ID"]}.json', "w") as fp:
            json.dump(assessment_dict, fp, indent=4)

        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.'
        )
        sys.exit(1)

1	"""Export all the data from an assessment within GoPhish into a single JSON file.
2
3	Usage:
4	gophish-export [--log-level=LEVEL] ASSESSMENT_ID SERVER API_KEY
5	gophish-export (-h \| --help)
6	gophish-export --version
7
8	Options:
9	API_KEY GoPhish API key.
10	ASSESSMENT_ID ID of the assessment to export data from.
11	SERVER Full URL to GoPhish server.
12	-h --help Show this screen.
13	--version Show version.
14	-l --log-level=LEVEL If specified, then the log level will be set to
15	the specified value. Valid values are "debug", "info",
16	"warning", "error", and "critical". [default: info]
17	"""
18
19	# Standard Python Libraries
20	from datetime import datetime	4✔
21	import hashlib	4✔
22	import json	4✔
23	import logging	4✔
24	import re	4✔
25	import sys	4✔
26	from typing import Dict	4✔
27
28	# Third-Party Libraries
29	from docopt import docopt	4✔
30	import httpagentparser	4✔
31	import requests	4✔
32
33	# cisagov Libraries
34	from tools.connect import connect_api	4✔
35
36	from ._version import __version__	4✔
37
38	# Disable "Insecure Request" warning: GoPhish uses a self-signed certificate
39	# as default for https connections, which can not be verified by a third
40	# party; thus, an SSL insecure request warning is produced.
41	requests.packages.urllib3.disable_warnings()	4✔
42
43
44	def assessment_exists(api, assessment_id):	4✔
45	"""Check if GoPhish has at least one campaign for designated assessment.
46
47	Args:
48	api (GoPhish API): Connection to GoPhish server via the API.
49	assessment_id (string): Assessment identifier to get campaigns from.
50
51	Returns:
52	boolean: Indicates if a campaign is found starting with assessment_id.
53	"""
54	allCampaigns = api.campaigns.get()	4✔
55	for campaign in allCampaigns:	4✔
56	if campaign.name.startswith(assessment_id):	4✔
57	return True	4✔
58
59	return False	4✔
60
61
62	def export_targets(api, assessment_id):	4✔
63	"""Add all targets to a list.
64
65	Achieved by pulling the group IDs for any group starting with
66	the assessment id. The targets within the group are then parsed
67	into a targets list of target dicts. Each target dict includes a
68	sha256 hash of the target's email and assessment id with any labels.
69
70	Args:
71	api (GoPhish API): Connection to GoPhish server via the API.
72	assessment_id (string): Assessment identifier to get campaigns from.
73
74	Returns:
75	List of targets from the assessment's group(s).
76	"""
77	groupIDs = get_group_ids(api, assessment_id)	4✔
78
79	targets = list()	4✔
80
81	for group_id in groupIDs:	4✔
82	# Gets target list for parsing.
83	raw_targets = api.groups.get(group_id).as_dict()["targets"]	4✔
84
85	for raw_target in raw_targets:	4✔
86
87	target = dict()	4✔
88
89	target["id"] = hashlib.sha256(	4✔
90	raw_target["email"].encode("utf-8")
91	).hexdigest()
92	target["customer_defined_labels"] = dict()	4✔
93
94	if "position" in raw_target:	4!
95	target["customer_defined_labels"][assessment_id] = [	4✔
96	raw_target["position"]
97	]
98
99	targets.append(target)	4✔
100
101	logging.info(f"{len(targets)} email targets found for assessment {assessment_id}.")	4✔
102
103	return targets	4✔
104
105
106	def get_group_ids(api, assessment_id):	4✔
107	"""Return a list of group IDs for all groups starting with specified assessment_id."""
108	rawGroup = api.groups.get() # Holds raw list of campaigns from GoPhish.	×
109	groups = list() # Holds list of campaign IDs that match the assessment.	×
110
111	for group in rawGroup:	×
112	group = group.as_dict()	×
113	if group["name"].startswith(assessment_id):	×
114	groups.append(group["id"])	×
115
116	return groups	×
117
118
119	def export_campaigns(api, assessment_id):	4✔
120	"""Add all the campaigns' data for an assessment to a list.
121
122	Args:
123	api (GoPhish API): Connection to GoPhish server via the API.
124	assessment_id (string): Assessment identifier to get campaigns from.
125
126	Returns:
127	List of the assessment's campaigns with data.
128	"""
129	campaignIDs = get_campaign_ids(api, assessment_id)	×
130	campaigns = list()	×
131
132	for campaign_id in campaignIDs:	×
133	campaigns.append(get_campaign_data(api, campaign_id))	×
134
135	logging.info(f"{len(campaigns)} campaigns found for assessment {assessment_id}.")	×
136
137	return campaigns	×
138
139
140	def get_campaign_ids(api, assessment_id):	4✔
141	"""Return a list of campaign IDs for all campaigns starting with specified assessment_id."""
142	rawCampaigns = api.campaigns.get() # Holds raw list of campaigns from GoPhish.	×
143	campaigns = list() # Holds list of campaign IDs that match the assessment.	×
144
145	for campaign in rawCampaigns:	×
146	campaign = campaign.as_dict()	×
147	if campaign["name"].startswith(assessment_id):	×
148	campaigns.append(campaign["id"])	×
149
150	return campaigns	×
151
152
153	def get_campaign_data(api, campaign_id):	4✔
154	"""Return campaign metadata for the given campaign ID."""
155	campaign = dict()	×
156
157	# Pulls the campaign data as dict from GoPhish.
158	rawCampaign: dict = api.campaigns.get(campaign_id).as_dict()	×
159
160	campaign["id"] = rawCampaign["name"]	×
161
162	campaign["start_time"] = rawCampaign["launch_date"]	×
163	campaign["end_time"] = rawCampaign["completed_date"]	×
164	campaign["url"] = rawCampaign["url"]	×
165
166	campaign["subject"] = rawCampaign["template"]["subject"]	×
167
168	# Get the template ID from the GoPhish template name.
169	campaign["template"] = (	×
170	api.templates.get(rawCampaign["template"]["id"]).as_dict()["name"].split("-")[2]
171	)
172
173	campaign["clicks"] = get_click_data(api, campaign_id)	×
174
175	# Get the e-mail send status from GoPhish.
176	campaign["status"] = get_email_status(api, campaign_id)	×
177
178	return campaign	×
179
180
181	def get_click_data(api, campaign_id):	4✔
182	"""Return a list of all clicks for a given campaign."""
183	rawEvents = api.campaigns.get(campaign_id).as_dict()["timeline"]	×
184	clicks = list() # Holds list of all users that clicked.	×
185
186	for rawEvent in rawEvents:	×
187	if rawEvent["message"] == "Clicked Link":	×
188	click = dict()	×
189
190	# Builds out click document.
191	click["user"] = hashlib.sha256(	×
192	rawEvent["email"].encode("utf-8")
193	).hexdigest()
194	click["source_ip"] = rawEvent["details"]["browser"]["address"]	×
195
196	click["time"] = rawEvent["time"]	×
197
198	click["application"] = get_application(rawEvent)	×
199
200	clicks.append(click)	×
201
202	return clicks	×
203
204
205	def get_email_status(api, campaign_id):	4✔
206	"""Return the email send status and time."""
207	rawEvents = api.campaigns.get(campaign_id).as_dict()["timeline"]	×
208	status = list()	×
209	for rawEvent in rawEvents:	×
210	email = dict()	×
211
212	if rawEvent["message"] == "Email Sent":	×
213	email["user"] = hashlib.sha256(	×
214	rawEvent["email"].encode("utf-8")
215	).hexdigest()
216
217	email["time"] = rawEvent["time"]	×
218
219	email["status"] = "SUCCESS"	×
220
221	elif rawEvent["message"] == "Error Sending Email":	×
222
223	email["user"] = hashlib.sha256(	×
224	rawEvent["email"].encode("utf-8")
225	).hexdigest()
226
227	# Trim microseconds before converting to datetime.
228	rawEvent["time"] = datetime.strptime(	×
229	rawEvent["time"].split(".")[0], "%Y-%m-%dT%H:%M:%S"
230	)
231	email["time"] = rawEvent["time"]	×
232
233	email["status"] = "Failed"	×
234
235	if email:	×
236	status.append(email)	×
237
238	return status	×
239
240
241	def get_application(rawEvent):	4✔
242	"""Return application details."""
243	application = dict()	×
244
245	application["external_ip"] = rawEvent["details"]["browser"]["address"]	×
246
247	# Process user agent string.
248	userAgent = rawEvent["details"]["browser"]["user-agent"]	×
249	application["name"] = httpagentparser.detect(userAgent)["platform"]["name"]	×
250	application["version"] = httpagentparser.detect(userAgent)["platform"]["version"]	×
251
252	return application	×
253
254
255	def find_unique_target_clicks_count(clicks):	4✔
256	"""Return the number of unique clicks in a click set."""
257	uniq_users = set()	4✔
258	for click in clicks:	4✔
259	uniq_users.add(click["user"])	4✔
260	return len(uniq_users)	4✔
261
262
263	def write_campaign_summary(api, assessment_id):	4✔
264	"""Output a campaign summary report to JSON, console, and a text file."""
NEW 265	campaign_ids = get_campaign_ids(api, assessment_id)	×
NEW 266	campaign_data_template = "campaign_data.json"	×
NEW 267	campaign_summary_json = f"{assessment_id}_campaign_data.json"	×
NEW 268	campaign_summary_textfile = f"{assessment_id}_summary_{datetime.strftime(datetime.now(), '%Y-%m-%dT%H:%M:%S')}.txt"	×
269
NEW 270	with open(campaign_data_template) as template:	×
NEW 271	campaign_data = json.load(template)	×
272
NEW 273	logging.info("Writing campaign summary report to %s", campaign_summary_textfile)	×
NEW 274	file_out = open(campaign_summary_textfile, "w+")	×
NEW 275	file_out.write("Campaigns for Assessment: " + assessment_id)	×
276
NEW 277	regex = re.compile(r"^.*_(?P<level>level-[1-6])$")	×
NEW 278	for campaign_id in campaign_ids:	×
NEW 279	campaign = api.campaigns.get(campaign_id)	×
NEW 280	match = regex.fullmatch(campaign.name)	×
NEW 281	if match:	×
NEW 282	level = match.group("level")	×
283	else:
NEW 284	logging.warn(	×
285	"Encountered campaign (%s) that is unable to be processed for campaign summary export. \n"
286	"Campaign name is not properly suffixed with the campaign level number (e.g. '_level-1')\n"
287	"Skipping campaign",
288	campaign.name,
289	)
NEW 290	continue	×
291
NEW 292	logging.info(level)	×
NEW 293	clicks = get_click_data(api, campaign_id)	×
294
NEW 295	total_clicks = api.campaigns.summary(campaign_id=campaign_id).stats.clicked	×
NEW 296	unique_clicks = find_unique_target_clicks_count(clicks)	×
NEW 297	if total_clicks > 0:	×
NEW 298	percent_clicks = unique_clicks / float(total_clicks)	×
299	else:
NEW 300	percent_clicks = 0.0	×
NEW 301	campaign_data[level]["subject"] = campaign.template.subject	×
NEW 302	campaign_data[level]["sender"] = campaign.smtp.from_address	×
NEW 303	campaign_data[level]["start_date"] = campaign.launch_date	×
NEW 304	campaign_data[level]["end_date"] = campaign.completed_date	×
NEW 305	campaign_data[level]["redirect"] = campaign.url	×
NEW 306	campaign_data[level]["clicks"] = total_clicks	×
NEW 307	campaign_data[level]["unique_clicks"] = unique_clicks	×
NEW 308	campaign_data[level]["percent_clicks"] = percent_clicks	×
309
NEW 310	file_out.write("\n")	×
NEW 311	file_out.write("-" * 50)	×
NEW 312	file_out.write("\nCampaign: %s" % campaign.name)	×
NEW 313	file_out.write("\nSubject: %s" % campaign_data[level]["subject"])	×
NEW 314	file_out.write("\nSender: %s" % campaign_data[level]["sender"])	×
NEW 315	file_out.write("\nStart Date: %s" % campaign_data[level]["start_date"])	×
NEW 316	file_out.write("\nEnd Date: %s" % campaign_data[level]["end_date"])	×
NEW 317	file_out.write("\nRedirect: %s" % campaign_data[level]["redirect"])	×
NEW 318	file_out.write("\nClicks: %i" % campaign_data[level]["clicks"])	×
NEW 319	file_out.write("\nUnique Clicks: %i" % campaign_data[level]["unique_clicks"])	×
NEW 320	file_out.write(	×
321	"\nPercentage Clicks: %f" % campaign_data[level]["percent_clicks"]
322	)
323
NEW 324	file_out.close()	×
NEW 325	logging.info("Writing out summary JSON to %s", campaign_summary_json)	×
NEW 326	with open(campaign_summary_json, "w") as fp:	×
NEW 327	json.dump(campaign_data, fp, indent=4)	×
328
329
330	def export_user_reports(api, assessment_id):	4✔
331	"""Build and export a user_report JSON file for each campaign in an assessment."""
332	campaign_ids = get_campaign_ids(api, assessment_id)	×
333
334	for campaign_id in campaign_ids:	×
335	first_report = None	×
336	user_report_doc = dict()	×
337	campaign = get_campaign_data(api, campaign_id)	×
338
339	# iterate over clicks and find the earliest click
340	for click in campaign["clicks"]:	×
341	click_time = datetime.strptime(	×
342	click["time"].split(".")[0], "%Y-%m-%dT%H:%M:%S"
343	)
344	if first_report is None or click_time < first_report:	×
345	first_report = click_time	×
346
347	# The "customer" field is a placeholder added for operator convenience when
348	# working with the JSON file created.
349	user_report_doc["customer"] = ""	×
350	user_report_doc["assessment"] = assessment_id	×
351	# get_campaign_ids() returns integers, but user_report_doc["campaign"]
352	# expects a string
353	user_report_doc["campaign"] = str(campaign_id)	×
NEW 354	if first_report is not None:	×
NEW 355	user_report_doc["first_report"] = datetime.strftime(	×
356	first_report, "%Y-%m-%dT%H:%M:%S"
357	)
358	else:
NEW 359	user_report_doc["first_report"] = "No clicks reported"	×
360
UNCOV 361	user_report_doc["total_num_reports"] = api.campaigns.summary(	×
362	campaign_id=campaign_id
363	).stats.clicked
364
365	logging.info(	×
366	f"Writing out user report for campaign {campaign_id} in assessment {assessment_id}"
367	)
368
369	with open(f"{assessment_id}_{campaign_id}_user_report_doc.json", "w") as fp:	×
370	json.dump(user_report_doc, fp, indent=4)	×
371
372
373	def main() -> None:	4✔
374	"""Set up logging, connect to API, export all assessment data."""
375	args: Dict[str, str] = docopt(__doc__, version=__version__)	×
376
377	# Set up logging
378	log_level = args["--log-level"]	×
379	try:	×
380	logging.basicConfig(	×
381	format="\n%(levelname)s: %(message)s", level=log_level.upper()
382	)
383	except ValueError:	×
384	logging.critical(	×
385	f'"{log_level}"is not a valid logging level. Possible values are debug, info, warning, and error.'
386	)
387	sys.exit(1)	×
388
389	else:
390	# Connect to API
391	try:	×
392	api = connect_api(args["API_KEY"], args["SERVER"])	×
393	logging.debug(f'Connected to: {args["SERVER"]}')	×
394	except Exception as e:	×
395	logging.critical(e.args[0])	×
396	sys.exit(1)	×
397
398	if assessment_exists(api, args["ASSESSMENT_ID"]):	×
399	assessment_dict: Dict = dict()	×
400
401	# Add targets list to assessment dict.
402	assessment_dict["targets"] = export_targets(api, args["ASSESSMENT_ID"])	×
403
404	# Add campaigns list to the assessment dict.
405	assessment_dict["campaigns"] = export_campaigns(api, args["ASSESSMENT_ID"])	×
406
407	with open(f'data_{args["ASSESSMENT_ID"]}.json', "w") as fp:	×
408	json.dump(assessment_dict, fp, indent=4)	×
409
410	logging.info(f'Data written to data_{args["ASSESSMENT_ID"]}.json')	×
411
412	export_user_reports(api, args["ASSESSMENT_ID"])	×
NEW 413	write_campaign_summary(api, args["ASSESSMENT_ID"])	×
414	else:
415	logging.error(	×
416	f'Assessment "{args["ASSESSMENT_ID"]}" does not exist in GoPhish.'
417	)
418	sys.exit(1)	×

cisagov / gophish-tools / 1218583375

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous