otrs-notification-public/otrs-notification.py

#!/usr/bin/python3

import json
import requests
import urllib.parse
from argparse import ArgumentParser
from pyotrs import Client, Article, Ticket

# ignore insecure SSL warnings
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# otrs connection information
OTRS = {
    "uri": "https://otrs2.inmedias.it/",
    "connector": "GenericTicketConnector",
    "username": "**********************",
    "password": "**********************"
}

NEW_TICKET_DEFAULTS = {
    "queue": "Monitoring",
    "priority": "2 normal",
    "owner": "admin",
    "state": "new",
    "time": 0
}

# icinga connection information
ICINGA = {
    "uri": "https://localhost:5665",
    "path": "/v1/actions/acknowledge-problem",
    "username": "********************",
    "password": "********************",
    "author": "admin"
}


def uri_encode(string):
    return urllib.parse.quote(string)


def get_wiki_entry(args):
    wiki_url = "https://wiki.inmedias.it/index.php?title=Monitoring:%s" % uri_encode(args.display_name)
    if args.type == "Service":
        wiki_url += "#%s" % uri_encode(args.service_name)
    return wiki_url.replace('%20', '_')


def get_icinga_entry(args):
    icinga_url = "https://deb-icinga2-proto.inmedias.it/icingaweb2/monitoring/%s/show?host=%s" % (args.type.lower(), uri_encode(args.hostname))
    if args.type == "Service":
        icinga_url += "&service=%s" % uri_encode(args.service_name)
    return icinga_url


def get_icinga_api_entry(args):
    if args.type == "Host":
        api_args = "?host=%s" % uri_encode(args.hostname)
    else:
        api_args = "?service=%s!%s" % (uri_encode(args.hostname), uri_encode(args.service_name))
    return api_args


def main(args):
    # initialize the client
    client = Client(OTRS['uri'], OTRS['username'], OTRS['password'])
    client.ws_ticket = OTRS['connector']
    client.session_create()

    # generate ticket attributes
    ticket_title = "[{status}] {hostname}"
    if args.type == "Service" or args.service_name:
        ticket_title += ": {service_name}"

    ticket_body = [
        "<pre>{}</pre>".format(args.details),
        '<a href="{ic_link}">{ic_title}</a>'.format(ic_link=get_icinga_entry(args), ic_title="Icinga Backlink"),
        '<a href="{wiki_link}">{wiki_title}</a>'.format(wiki_link=get_wiki_entry(args), wiki_title="Wiki Backlink")
    ]

    # # OTRS API Documentation:
    # # http://doc.otrs.com/doc/api/otrs/6.0/Perl/Kernel/GenericInterface/Operation/Ticket/TicketCreate.pm.html

    # create first article for new ticket
    article = Article({
        "Subject": ticket_title.format(**args.__dict__),
        "Body": "<br>".join(ticket_body),
        "TimeUnit": args.time,
        "MimeType": "text/html",
        "NoAgentNotify": int(args.no_agent_notify)
    })

    # create ticket with article
    new_ticket = Ticket.create_basic(
        Title=ticket_title.format(**args.__dict__),
        Queue=args.queue,
        State=args.state,
        Priority=args.priority,
        CustomerUser=args.customer
    )
    ticket = client.ticket_create(ticket=new_ticket, article=article)

    # set ticket owner
    client.ticket_update(ticket_id=ticket.get("TicketID"), Owner=args.owner)

    # set Acknowledgement in Icinga
    ticket_number = ticket.get("TicketNumber")
    ack = requests.post(url="{uri}{path}{args}".format(**ICINGA, args=get_icinga_api_entry(args)),
                        data=json.dumps({
                            "author": ICINGA['author'],
                            "comment": "Ticket erstellt #%s" % ticket_number,
                            "sticky": 1
                        }),
                        headers={"Accept": "application/json"},
                        auth=(ICINGA['username'], ICINGA['password']),
                        verify=False)
    if ack.ok:
        # exit successfully
        print("Done.")
        exit(0)
    else:
        raise Exception("Could not acknowlede ticket '%s' in Icinga. Response: \n%s" % (ticket_number, ack.__dict__))


if __name__ == "__main__":
    # parse arguments taking into account default values
    d_args = NEW_TICKET_DEFAULTS
    parser = ArgumentParser(description='Notification Command for OTRS.', conflict_handler='resolve')

    parser.add_argument('-t', '--type', help='Check type: either `Host` (default) or `Service`', default="Host")
    parser.add_argument('-H', '--hostname', help='Hostname (FQDN) or IP address as seen in Icinga.', required=True)
    parser.add_argument('-D', '--display-name', help='Display Name of Host as seen in Icinga.', required=True)
    parser.add_argument('-N', '--service-name', help='Service name as seen in Icinga.')
    parser.add_argument('-d', '--details', '--description', help='Check description/details', required=True)
    parser.add_argument('-s', '--status', help='Check status: WARN, CRIT, UNKNOWN', required=True)
    parser.add_argument('-c', '--customer', help='Customer user for which to create ticket in OTRS', required=True)

    parser.add_argument('-P', '--priority', help='OTRS ticket priority. Defaults to `{priority}`'.format(**d_args),
                        default=d_args['priority'])
    parser.add_argument('-O', '--owner', help='OTRS ticket owner. Defaults to `{owner}`'.format(**d_args),
                        default=d_args['owner'])
    parser.add_argument('-S', '--state', help='OTRS ticket state. Defaults to `{state}`'.format(**d_args),
                        default=d_args['state'])
    parser.add_argument('-Q', '--queue', help='OTRS ticket queue. Defaults to `{queue}`'.format(**d_args),
                        default=d_args['queue'])
    parser.add_argument('-T', '--time',
                        help='Time in minutes to book into new ticket in OTRS. Defaults to `{time}`'.format(**d_args),
                        type=int, default=d_args['time'])

    parser.add_argument('--no-agent-notify', help="Whether to notify the agent about new ticket.", action='store_true')

    # run main with parsed arguments
    main(parser.parse_args())