4693799403

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #123 from cisagov/lineage/skeleton

Run Details

141 of 473 branches covered (29.81%)

Branch coverage included in aggregate %.

9 of 24 new or added lines in 10 files covered. (37.5%)

223 existing lines in 5 files now uncovered.

298 of 1270 relevant lines covered (23.46%)

1.41 hits per line

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

22.86

/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
import urllib3

# 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.
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(
            '"%s" is not a valid logging level. Possible values are debug, info, warning, and error.',
            log_level,
        )
        sys.exit(1)

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

cisagov / gophish-tools / 4693799403

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