1176040445

Build Type

github

Committed by GitHub

Commit Message

Merge 097b9c665 into 22ccab80f

Pull Request Pull Request #40: Add campaign summary export

Run Details

94 of 392 branches covered (23.98%)

Branch coverage included in aggregate %.

15 of 80 new or added lines in 3 files covered. (18.75%)

229 existing lines in 8 files now uncovered.

292 of 1266 relevant lines covered (23.06%)

0.92 hits per line

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

22.14

/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 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 dict."""
    uniq_list = list()
    for click in clicks:
        if click["user"] not in uniq_list:
            uniq_list.append(click["user"])
    return len(uniq_list)


def write_campaign_summary(api, assessment_id):
    """Output a campaign summary report to JSON, console, and a text file."""
    campaign_data = dict()
    campaign_ids = get_campaign_ids(api, assessment_id)
    campaign_data_template = "campaign_data.json"
    campaign_summary_json = assessment_id + "_campaign_data.json"
    campaign_summary_textfile = (
        assessment_id + "_summary_" + str(datetime.now()) + ".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)

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

cisagov / gophish-tools / 1176040445

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