1151693420

Build Type

github

Committed by GitHub

Commit Message

Merge c0e7d5868 into 44e0cfb34

Pull Request Pull Request #40: Add campaign summary export

Run Details

94 of 382 branches covered (24.61%)

Branch coverage included in aggregate %.

16 of 60 new or added lines in 3 files covered. (26.67%)

241 existing lines in 8 files now uncovered.

293 of 1249 relevant lines covered (23.46%)

0.94 hits per line

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

22.12

/src/tools/gophish_complete.py

"""Complete a campaign in GoPhish and/or output a GoPhish campaign summary.

Usage:
  gophish-complete [--campaign=NAME] [--summary-only] [--log-level=LEVEL] SERVER API_KEY
  gophish-complete (-h | --help)
  gophish-complete --version

Options:
  API_KEY                   GoPhish API key.
  SERVER                    Full URL to GoPhish server.
  -c --campaign=NAME        GoPhish campaign name.
  -s --summary-only         Output a summary of a GoPhish campaign.
  -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]

NOTE:
  * If a campaign name is not provided, all assessment campaigns will be listed to select from.
"""

# import IPython; IPython.embed() #<<< BREAKPOINT >>>
# sys.exit(0)

# Standard Python Libraries
import logging
import sys
from typing import Dict

# Third-Party Libraries
from docopt import docopt
import requests

# cisagov Libraries
from tools.connect import connect_api
from util.input import get_input, get_number

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 get_campaign_id(campaign_name, campaigns):
    """Get campaign id from campaign name.

    Args:
        campaign_name (string): Full campaign name.
        campaigns (dict): Campaign id as key, campaign name as value.

    Raises:
        LookupError: Campaign name is not found in campaigns dictionary.

    Returns:
        Campaign id corresponding to the campaign name provided.
    """
    for campaign_id, name_value in campaigns.items():
        if name_value == campaign_name:
            return campaign_id

    raise LookupError(f'Campaign name "{campaign_name}" not found.')


def get_campaigns(api, assessment_id=""):
    """Return a dictionary containing all campaigns.

    When called with a blank string for the assessment_id, the default value,
    all campaigns in all assessments will be returned. If an assessment_id is
    provided, then only the campaigns for that assessment will be returned.

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

    Raises:
        LookupError: No campaigns found for the provided assessment id.

    Returns:
        dict: Campaign id as key, campaign name as value.
    """
    allCampaigns = api.campaigns.get()

    assessmentCampaigns = dict()

    for campaign in allCampaigns:
        if campaign.name.startswith(assessment_id):
            assessmentCampaigns[campaign.id] = campaign.name

    if len(assessmentCampaigns) == 0:
        raise LookupError(f"No campaigns found for assessment {assessment_id}")

    return assessmentCampaigns


def select_campaign(campaigns):
    """Return the ID of a selected campaign."""
    print("Please select a Campaign ID:")
    print("\tID: Name")

    for id, name in campaigns.items():
        print(f"\t {id}: {name}")

    print("")

    while True:
        inputId = get_number("ID: ")
        if inputId in campaigns:
            break
        else:
            logging.warning("Bad Campaign ID")
            print("Try again...")

    return inputId


def complete_campaign(api_key, server, campaign_id):
    """Complete a campaign in GoPhish.

    Args:
        api_key (string): GoPhish API key.
        server (string): Full URL to GoPhish server.
        campaign_id (int): GoPhish campaign id.

    Raises:
        UserWarning: GoPhish is unsuccessful in completing the campaign.
    """
    url = f"{server}/api/campaigns/{campaign_id}/complete?api_key={api_key}"

    # Bandit complains about disabling the SSL certificate check, but we have
    # no choice here since we are using a self-signed certificate.
    response = requests.get(url=url, verify=False)  # nosec

    if not response.json()["success"]:
        raise UserWarning(response.json()["message"])
    else:
        print(f'\n{response.json()["message"]}')


def print_summary(api, campaign_id):
    """Print a campaign summary."""
    summary = api.campaigns.summary(campaign_id=campaign_id)

    print("Campaign Summary:")
    print(f"\tName: {summary.name}")
    print(f"\tStatus: {summary.status}")
    print(f"\tLaunch Date: {summary.launch_date}")
    print(f"\tCompleted Date: {summary.completed_date}")
    print(f"\tTotal Users: {summary.stats.total}")
    print(f"\tTotal Sent: {summary.stats.sent}")
    print(f"\tTotal Clicks:  {summary.stats.clicked}")

    return True


def main() -> None:
    """Set up logging, connect to API, call requested function(s)."""
    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)

    # 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)

    try:
        if args["--campaign"]:
            # Use campaign name to find campaign id.
            campaigns = get_campaigns(api)
            campaign_id = get_campaign_id(args["--campaign"], campaigns)
        else:
            # User inputs assessment id and selects campaign from lists.
            assessment_id = get_input("Enter the Assessment ID:")
            campaigns = get_campaigns(api, assessment_id)
            campaign_id = select_campaign(campaigns)

    except LookupError as err:
        logging.error(err)
        sys.exit(1)

    if args["--summary-only"]:
        # Output summary only.
        print_summary(api, campaign_id)
    else:
        # Complete and output summary.
        try:
            complete_campaign(args["API_KEY"], args["SERVER"], campaign_id)

        except UserWarning as err:
            logging.warning(err)
            sys.exit(1)

        print_summary(api, campaign_id)

1	"""Complete a campaign in GoPhish and/or output a GoPhish campaign summary.
2
3	Usage:
4	gophish-complete [--campaign=NAME] [--summary-only] [--log-level=LEVEL] SERVER API_KEY
5	gophish-complete (-h \| --help)
6	gophish-complete --version
7
8	Options:
9	API_KEY GoPhish API key.
10	SERVER Full URL to GoPhish server.
11	-c --campaign=NAME GoPhish campaign name.
12	-s --summary-only Output a summary of a GoPhish campaign.
13	-h --help Show this screen.
14	--version Show version.
15	-l --log-level=LEVEL If specified, then the log level will be set to
16	the specified value. Valid values are "debug", "info",
17	"warning", "error", and "critical". [default: info]
18
19	NOTE:
20	* If a campaign name is not provided, all assessment campaigns will be listed to select from.
21	"""
22
23	# import IPython; IPython.embed() #<<< BREAKPOINT >>>
24	# sys.exit(0)
25
26	# Standard Python Libraries
27	import logging	4✔
28	import sys	4✔
29	from typing import Dict	4✔
30
31	# Third-Party Libraries
32	from docopt import docopt	4✔
33	import requests	4✔
34
35	# cisagov Libraries
36	from tools.connect import connect_api	4✔
37	from util.input import get_input, get_number	4✔
38
39	from ._version import __version__	4✔
40
41	# Disable "Insecure Request" warning: GoPhish uses a self-signed certificate
42	# as default for https connections, which can not be verified by a third
43	# party; thus, an SSL insecure request warning is produced.
44	requests.packages.urllib3.disable_warnings()	4✔
45
46
47	def get_campaign_id(campaign_name, campaigns):	4✔
48	"""Get campaign id from campaign name.
49
50	Args:
51	campaign_name (string): Full campaign name.
52	campaigns (dict): Campaign id as key, campaign name as value.
53
54	Raises:
55	LookupError: Campaign name is not found in campaigns dictionary.
56
57	Returns:
58	Campaign id corresponding to the campaign name provided.
59	"""
60	for campaign_id, name_value in campaigns.items():	4✔
61	if name_value == campaign_name:	4✔
62	return campaign_id	4✔
63
64	raise LookupError(f'Campaign name "{campaign_name}" not found.')	4✔
65
66
67	def get_campaigns(api, assessment_id=""):	4✔
68	"""Return a dictionary containing all campaigns.
69
70	When called with a blank string for the assessment_id, the default value,
71	all campaigns in all assessments will be returned. If an assessment_id is
72	provided, then only the campaigns for that assessment will be returned.
73
74	Args:
75	api (GoPhish API): Connection to GoPhish server via the API.
76	assessment_id (string): Assessment identifier to get campaigns from.
77
78	Raises:
79	LookupError: No campaigns found for the provided assessment id.
80
81	Returns:
82	dict: Campaign id as key, campaign name as value.
83	"""
UNCOV 84	allCampaigns = api.campaigns.get()	×
85
86	assessmentCampaigns = dict()	×
87
88	for campaign in allCampaigns:	×
UNCOV 89	if campaign.name.startswith(assessment_id):	×
90	assessmentCampaigns[campaign.id] = campaign.name	×
91
92	if len(assessmentCampaigns) == 0:	×
UNCOV 93	raise LookupError(f"No campaigns found for assessment {assessment_id}")	×
94
95	return assessmentCampaigns	×
96
97
98	def select_campaign(campaigns):	4✔
99	"""Return the ID of a selected campaign."""
UNCOV 100	print("Please select a Campaign ID:")	×
UNCOV 101	print("\tID: Name")	×
102
103	for id, name in campaigns.items():	×
UNCOV 104	print(f"\t {id}: {name}")	×
105
106	print("")	×
107
108	while True:
UNCOV 109	inputId = get_number("ID: ")	×
UNCOV 110	if inputId in campaigns:	×
111	break	×
112	else:
113	logging.warning("Bad Campaign ID")	×
UNCOV 114	print("Try again...")	×
115
116	return inputId	×
117
118
119	def complete_campaign(api_key, server, campaign_id):	4✔
120	"""Complete a campaign in GoPhish.
121
122	Args:
123	api_key (string): GoPhish API key.
124	server (string): Full URL to GoPhish server.
125	campaign_id (int): GoPhish campaign id.
126
127	Raises:
128	UserWarning: GoPhish is unsuccessful in completing the campaign.
129	"""
UNCOV 130	url = f"{server}/api/campaigns/{campaign_id}/complete?api_key={api_key}"	×
131
132	# Bandit complains about disabling the SSL certificate check, but we have
133	# no choice here since we are using a self-signed certificate.
UNCOV 134	response = requests.get(url=url, verify=False) # nosec	×
135
136	if not response.json()["success"]:	×
UNCOV 137	raise UserWarning(response.json()["message"])	×
138	else:
139	print(f'\n{response.json()["message"]}')	×
140
141
142	def print_summary(api, campaign_id):	4✔
143	"""Print a campaign summary."""
UNCOV 144	summary = api.campaigns.summary(campaign_id=campaign_id)	×
145
146	print("Campaign Summary:")	×
UNCOV 147	print(f"\tName: {summary.name}")	×
148	print(f"\tStatus: {summary.status}")	×
149	print(f"\tLaunch Date: {summary.launch_date}")	×
150	print(f"\tCompleted Date: {summary.completed_date}")	×
151	print(f"\tTotal Users: {summary.stats.total}")	×
152	print(f"\tTotal Sent: {summary.stats.sent}")	×
153	print(f"\tTotal Clicks: {summary.stats.clicked}")	×
154
155	return True	×
156
157
158	def main() -> None:	4✔
159	"""Set up logging, connect to API, call requested function(s)."""
UNCOV 160	args: Dict[str, str] = docopt(__doc__, version=__version__)	×
161
162	# Set up logging
UNCOV 163	log_level = args["--log-level"]	×
UNCOV 164	try:	×
165	logging.basicConfig(	×
166	format="\n%(levelname)s: %(message)s", level=log_level.upper()
167	)
UNCOV 168	except ValueError:	×
UNCOV 169	logging.critical(	×
170	f'"{log_level}"is not a valid logging level. Possible values are debug, info, warning, and error.'
171	)
UNCOV 172	sys.exit(1)	×
173
174	# Connect to API
UNCOV 175	try:	×
UNCOV 176	api = connect_api(args["API_KEY"], args["SERVER"])	×
177	logging.debug(f'Connected to: {args["SERVER"]}')	×
178	except Exception as e:	×
179	logging.critical(e.args[0])	×
180	sys.exit(1)	×
181
182	try:	×
UNCOV 183	if args["--campaign"]:	×
184	# Use campaign name to find campaign id.
185	campaigns = get_campaigns(api)	×
UNCOV 186	campaign_id = get_campaign_id(args["--campaign"], campaigns)	×
187	else:
188	# User inputs assessment id and selects campaign from lists.
UNCOV 189	assessment_id = get_input("Enter the Assessment ID:")	×
UNCOV 190	campaigns = get_campaigns(api, assessment_id)	×
191	campaign_id = select_campaign(campaigns)	×
192
193	except LookupError as err:	×
UNCOV 194	logging.error(err)	×
195	sys.exit(1)	×
196
197	if args["--summary-only"]:	×
198	# Output summary only.
199	print_summary(api, campaign_id)	×
200	else:
201	# Complete and output summary.
UNCOV 202	try:	×
UNCOV 203	complete_campaign(args["API_KEY"], args["SERVER"], campaign_id)	×
204
205	except UserWarning as err:	×
UNCOV 206	logging.warning(err)	×
207	sys.exit(1)	×
208
209	print_summary(api, campaign_id)	×

cisagov / gophish-tools / 1151693420

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