files: Autostarting Security Monkey
  ===============================================
  
  This document outlines how to configure Security Monkey to:
  
  1. Automatically run the API
  1. Automatically scan for changes in your environment.
  1. Configure Security Monkey to send scanning performance metrics
  
  Each section is important, please read them thoroughly.
  
  Celery and The Supervisor
  ------------
  Security Monkey makes use of the excellent [Celery Distributed Task Queue for Python](http://docs.celeryproject.org/en/latest/index.html),
  and [Supervisord](http://supervisord.org/) (hence referred to as the Supervisor) for automatically starting the required components
  to make Security Monkey function.
  
  In general:
  1. The Supervisor ensures that there is always a Security Monkey related process running.
  1. Celery is used to manage the scheduling of worker instances to fetch and audit details
     of your accounts.
  
  For each, a sample Supervisor configuration is included. More details is provided in each section below.
  
  üö®‚ö†Ô∏èü•Åüé∫ IMPORTANT PLEASE READ üé∫ü•Å‚ö†Ô∏èüö®
  -------------
  The recommended deployment strategy of Security Monkey is to have 3 types of instances:
  1. 1 or many UI instances behind a load balancer
  1. Exactly ONE "scheduler" instance (more on that below)
  1. 1 or many "worker" instances that fetch details about the environment.
  1. A Redis cache (recommended -- it's easy and works great) to act as a message broker for the scheduler and workers to orchestrate task executions.
  
  *DO NOT* mix the "scheduler" and "worker" instances together -- unless you only want 1 instance doing everything
  (not recommended -- it doesn't scale).
  
  Each function should exist on separate instances. If you followed the quickstart guide, now is a good time
  to make an image of your instance/container and clone new ones to provide the separate configurations.
  
  For AWS, you can make an AMI off of your Security Monkey instance, and even create separate launch
  configurations and auto-scaling groups.
  
  Below is a diagram of the components required by Security Monkey (these are not AWS specific):
  ![diagram](images/sm_instance_diagram.png)
  
  Autostarting the API and UI
  --------------------
  
  The UI and API go hand-in-hand. The UI is started by NGINX, and the API is autostarted by the Supervisor.
  
  All sample supervisor configurations are stored in the [`supervisor`](https://github.com/Netflix/security_monkey/tree/develop/supervisor)
  location. For the UI, you will care about `security_monkey_ui.conf`(https://github.com/Netflix/security_monkey/tree/develop/supervisor/security_monkey_ui.conf).
  
  This file needs to be copied over to the supervisor configuration directory at `/etc/supervisor/conf.d/`.
  
  To enable NGINX and the supervisor, run the following commands (these also set the proper permissions):
  
      sudo chgrp -R www-data /var/log/security_monkey
      sudo cp /usr/local/src/security_monkey/supervisor/security_monkey_ui.conf /etc/supervisor/conf.d/security_monkey_ui.conf
      sudo systemctl enable nginx
      sudo systemctl enable supervisor
      sudo systemctl start nginx
      sudo systemctl start supervisor
      sudo supervisorctl status
  
  The `supervisorctl status` should output details about the loaded supervisor job.
  
  *Note:* If you make a change to a supervisor configuration, you will need to reload the supervisor service:
  `sudo systemctl reload supervisor`
  
  The supervisor will run the Security Monkey's Python [gunicorn](http://gunicorn.org/) command that launches the API (which runs the `monkey run_api_server` command.).
  
  You can track progress by tailing `/var/log/security_monkey/securitymonkey.log`.
  
  The Scheduler
  --------------
  The scheduler is used to configure the Celery workers. This schedules jobs onto a queue (message broker) which the Celery workers
  then use to fetch the tasks to work on in the proper intervals.
  
  ### The Message Broker (queue)
  Celery has several options for configuring a message broker, which is [documented in detail here](http://docs.celeryproject.org/en/latest/getting-started/brokers/).
  
  We recommend [Redis](https://redis.io/) as it's simple to configure. For AWS, recommend ElastiCache with Redis. We have instructions for setting up
  [ElastiCache instructions here](elasticache_directions.md).
  
  Security Monkey by default will install the `redis`, and `celery[redis]` packages. If you wish to install
  a different broker, you will need to follow the Celery documentation to install the proper dependencies.
  
  If you wish to use Redis on `localhost` (scheduler instance only), firewall rules need to be established to permit the worker
  instances access to it. An external Redis cache will also need to have
  firewall rules to permit worker instances access. This is documented in the above instructions for ElastiCache.
  
  If installing on the `localhost` of the scheduler instance, you will need to install Redis on the instance (this is complete if following the quickstart guide).
  
  ### Celery Configuration
  You will need to modify the `security_monkey/celeryconfig.py` file that is stored in the base Security Monkey location
  at `/usr/local/src/security_monkey/security_monkey/celeryconfig.py`(https://github.com/Netflix/security_monkey/blob/develop/security_monkey/celeryconfig.py).
  
  This file looks like this:
  ```
  # Broker source: Place yours here:
  broker_url = 'redis://localhost:6379/0'     # Default assume Redis on localhost
  
  # List of modules to import when the Celery worker starts.
  imports = ('security_monkey.task_scheduler.tasks',)
  
  # How many processes per worker instance?
  worker_concurrency = 10
  
  timezone = "UTC"
  enable_utc = True
  
  ###########################
  # IMPORTANT: This helps avoid memory leak issues - do not change this number!
  worker_max_tasks_per_child = 1
  ############################
  ```
  
  Only modify the `broker_url` path to point to the path to the broker you use. The default is
  configured to a `localhost` Redis.
  
  You may optionally adjust the `worker_concurrency`, which set the number of processes on an instance that the workers will
  use to fetch and audit your accounts.
  
  ### Supervisor
  With the message broker configured, it's now time to copy over the supervisor configuration.
  A sample is provided at: `security_monkey_scheduler.conf`(https://github.com/Netflix/security_monkey/tree/develop/supervisor/security_monkey_scheduler.conf)
  
  Run the following commands to set this up:
  
      sudo chgrp -R www-data /var/log/security_monkey
      sudo cp /usr/local/src/security_monkey/supervisor/security_monkey_scheduler.conf /etc/supervisor/conf.d/security_monkey_scheduler.conf
      sudo systemctl enable supervisor
      sudo systemctl start supervisor
      sudo supervisorctl status
  
  The supervisor configuration will start the Celery `beat` service, which performs all the scheduling logic. 
  The full command that the supervisor runs to launch the the `beat` service is:
   ```
   /usr/local/src/security_monkey/venv/bin/celery -A security_monkey.task_scheduler.beat.CELERY beat -s /var/run/sm-celerybeat-schedule -l debug
   ```
  
  Running this command will first purge out all existing jobs from the message broker and start fresh. This is
  necessary when editing watcher frequency and enablement.
  
  *Note*: There should be exactly ONE instance that runs `beat`. Running more than one will cause issues. 
  See deployment strategy below for additional details.
  
  The Workers
  ----------
  The workers are instances that fetch data from your configured accounts. These are instances that will:
  1. Fetch changes in your accounts
  1. Audit the changes
  1. Send out reports
  
  You are able to deploy as many workers as you like for your environment.  Security Monkey splits up tasks based on the account and technology pair.
  
  Similar to configuring the Scheduler above, the workers need to have the **same** `security_monkey/celeryconfig.py` as the scheduler. In here, you can optionally configure
  the number of processes that exist within each instance (via the `worker_concurrency` configuration). By default 10 is selected. You can adjust this as necessary. In general, if you would like to 
  scale horizontally, you should deploy more worker instances. This will allow for maximum parallelization.
  
  ### Supervisor
  
  A sample Supervisor configuration is provided at `security_monkey_workers.conf`(https://github.com/Netflix/security_monkey/tree/develop/supervisor/security_monkey_workers.conf)
  that will automatically run the Celery worker processes for you.
  
  Run the following commands to set this up:
  
      sudo chgrp -R www-data /var/log/security_monkey
      sudo cp /usr/local/src/security_monkey/supervisor/security_monkey_workers.conf /etc/supervisor/conf.d/security_monkey_workers.conf
      sudo systemctl enable supervisor
      sudo systemctl start supervisor
      sudo supervisorctl status
  
  
  Supervisor will run the Celery `worker` command, which is:
  ```
   /usr/local/src/security_monkey/venv/bin/celery -A security_monkey.task_scheduler.tasks.CELERY worker
  ```
  
  **Note:** Please do not run this on the same instance as the scheduler. Exactly one scheduler should exist. There can be many worker instances,
  so keep the supervisor configurations on these instances separate.
  
  
  Configure Security Monkey to send scanning performance metrics
  --------------------------------------------------------------
  Security Monkey can be configured to send metrics when objects are added or removed from the scanning queue.  This allows operators to check Security Monkey performance and ensure that items are being processed from the queue in a timely manner.  To do so set `METRICS_ENABLED` to `True`.  You will need `cloudwatch:PutMetricData` permission.  Metrics will be posted with the namespace `securitymonkey` unless configured using the variable `METRICS_NAMESPACE`.  You will also want to set `METRICS_POST_REGION` with the region you want to post CloudWatch Metrics to (default: `us-east-1`).
  
  
  Deployment Strategies
  --------------------
  A typical deployment strategy is:
  1. Deploy the UI behind a Load Balancer
  1. Tear-down the scheduler instance
  1. Tear-down all worker instances
  1. *At this point, you should perform any database migrations and upgrades as there are no workers that will be affected.*
  1. Deploy the new scheduler instance -- wait for it to come online
  1. Deploy many worker instances
  
  Performing the steps in this order will ensure:
  - That no duplicate schedulers are running (multiple schedulers running causes havoc)
  - Proper DB upgrades occur without possibly impacting workers mutating the database
  - The schedulers and workers are working properly together
  
files: docs/autostarting.md
files: Autostarting Security Monkey
  ===============================================
  
  This document outlines how to configure Security Monkey to:
  
  1. Automatically run the API
  1. Automatically scan for changes in your environment.
  
  Each section is important, please read them thoroughly.
  
  Celery and The Supervisor
  ------------
  Security Monkey makes use of the excellent [Celery Distributed Task Queue for Python](http://docs.celeryproject.org/en/latest/index.html),
  and [Supervisord](http://supervisord.org/) (hence referred to as the Supervisor) for automatically starting the required components
  to make Security Monkey function.
  
  In general:
  1. The Supervisor ensures that there is always a Security Monkey related process running.
  1. Celery is used to manage the scheduling of worker instances to fetch and audit details
     of your accounts.
  
  For each, a sample Supervisor configuration is included. More details is provided in each section below.
  
  üö®‚ö†Ô∏èü•Åüé∫ IMPORTANT PLEASE READ üé∫ü•Å‚ö†Ô∏èüö®
  -------------
  The recommended deployment strategy of Security Monkey is to have 3 types of instances:
  1. 1 or many UI instances behind a load balancer
  1. Exactly ONE "scheduler" instance (more on that below)
  1. 1 or many "worker" instances that fetch details about the environment.
  1. A Redis cache (recommended -- it's easy and works great) to act as a message broker for the scheduler and workers to orchestrate task executions.
  
  *DO NOT* mix the "scheduler" and "worker" instances together -- unless you only want 1 instance doing everything
  (not recommended -- it doesn't scale).
  
  Each function should exist on separate instances. If you followed the quickstart guide, now is a good time
  to make an image of your instance/container and clone new ones to provide the separate configurations.
  
  For AWS, you can make an AMI off of your Security Monkey instance, and even create separate launch
  configurations and auto-scaling groups.
  
  Below is a diagram of the components required by Security Monkey (these are not AWS specific):
  ![diagram](images/sm_instance_diagram.png)
  
  Autostarting the API and UI
  --------------------
  
  The UI and API go hand-in-hand. The UI is started by NGINX, and the API is autostarted by the Supervisor.
  
  All sample supervisor configurations are stored in the [`supervisor`](https://github.com/Netflix/security_monkey/tree/develop/supervisor)
  location. For the UI, you will care about `security_monkey_ui.conf`(https://github.com/Netflix/security_monkey/tree/develop/supervisor/security_monkey_ui.conf).
  
  This file needs to be copied over to the supervisor configuration directory at `/etc/supervisor/conf.d/`.
  
  To enable NGINX and the supervisor, run the following commands (these also set the proper permissions):
  
      sudo chgrp -R www-data /var/log/security_monkey
      sudo cp /usr/local/src/security_monkey/supervisor/security_monkey_ui.conf /etc/supervisor/conf.d/security_monkey_ui.conf
      sudo systemctl enable nginx
      sudo systemctl enable supervisor
      sudo systemctl start nginx
      sudo systemctl start supervisor
      sudo supervisorctl status
  
  The `supervisorctl status` should output details about the loaded supervisor job.
  
  *Note:* If you make a change to a supervisor configuration, you will need to reload the supervisor service:
  `sudo systemctl reload supervisor`
  
  The supervisor will run the Security Monkey's Python [gunicorn](http://gunicorn.org/) command that launches the API (which runs the `monkey run_api_server` command.).
  
  You can track progress by tailing `/var/log/security_monkey/securitymonkey.log`.
  
  The Scheduler
  --------------
  The scheduler is used to configure the Celery workers. This schedules jobs onto a queue (message broker) which the Celery workers
  then use to fetch the tasks to work on in the proper intervals.
  
  ### The Message Broker (queue)
  Celery has several options for configuring a message broker, which is [documented in detail here](http://docs.celeryproject.org/en/latest/getting-started/brokers/).
  
  We recommend [Redis](https://redis.io/) as it's simple to configure. For AWS, recommend ElastiCache with Redis. We have instructions for setting up
  [ElastiCache instructions here](elasticache_directions.md).
  
  Security Monkey by default will install the `redis`, and `celery[redis]` packages. If you wish to install
  a different broker, you will need to follow the Celery documentation to install the proper dependencies.
  
  If you wish to use Redis on `localhost` (scheduler instance only), firewall rules need to be established to permit the worker
  instances access to it. An external Redis cache will also need to have
  firewall rules to permit worker instances access. This is documented in the above instructions for ElastiCache.
  
  If installing on the `localhost` of the scheduler instance, you will need to install Redis on the instance (this is complete if following the quickstart guide).
  
  ### Celery Configuration
  You will need to modify the `security_monkey/celeryconfig.py` file that is stored in the base Security Monkey location
  at `/usr/local/src/security_monkey/security_monkey/celeryconfig.py`(https://github.com/Netflix/security_monkey/blob/develop/security_monkey/celeryconfig.py).
  
  This file looks like this:
  ```
  # Broker source: Place yours here:
  broker_url = 'redis://localhost:6379/0'     # Default assume Redis on localhost
  
  # List of modules to import when the Celery worker starts.
  imports = ('security_monkey.task_scheduler.tasks',)
  
  # How many processes per worker instance?
  worker_concurrency = 10
  
  timezone = "UTC"
  enable_utc = True
  
  ###########################
  # IMPORTANT: This helps avoid memory leak issues - do not change this number!
  worker_max_tasks_per_child = 1
  ############################
  ```
  
  Only modify the `broker_url` path to point to the path to the broker you use. The default is
  configured to a `localhost` Redis.
  
  You may optionally adjust the `worker_concurrency`, which set the number of processes on an instance that the workers will
  use to fetch and audit your accounts.
  
  ### Supervisor
  With the message broker configured, it's now time to copy over the supervisor configuration.
  A sample is provided at: `security_monkey_scheduler.conf`(https://github.com/Netflix/security_monkey/tree/develop/supervisor/security_monkey_scheduler.conf)
  
  Run the following commands to set this up:
  
      sudo chgrp -R www-data /var/log/security_monkey
      sudo cp /usr/local/src/security_monkey/supervisor/security_monkey_scheduler.conf /etc/supervisor/conf.d/security_monkey_scheduler.conf
      sudo systemctl enable supervisor
      sudo systemctl start supervisor
      sudo supervisorctl status
  
  The supervisor configuration will start the Celery `beat` service, which performs all the scheduling logic. 
  The full command that the supervisor runs to launch the the `beat` service is:
   ```
   /usr/local/src/security_monkey/venv/bin/celery -A security_monkey.task_scheduler.beat.CELERY beat -s /var/run/sm-celerybeat-schedule -l debug
   ```
  
  Running this command will first purge out all existing jobs from the message broker and start fresh. This is
  necessary when editing watcher frequency and enablement.
  
  *Note*: There should be exactly ONE instance that runs `beat`. Running more than one will cause issues. 
  See deployment strategy below for additional details.
  
  The Workers
  ----------
  The workers are instances that fetch data from your configured accounts. These are instances that will:
  1. Fetch changes in your accounts
  1. Audit the changes
  1. Send out reports
  
  You are able to deploy as many workers as you like for your environment.  Security Monkey splits up tasks based on the account and technology pair.
  
  Similar to configuring the Scheduler above, the workers need to have the **same** `security_monkey/celeryconfig.py` as the scheduler. In here, you can optionally configure
  the number of processes that exist within each instance (via the `worker_concurrency` configuration). By default 10 is selected. You can adjust this as necessary. In general, if you would like to 
  scale horizontally, you should deploy more worker instances. This will allow for maximum parallelization.
  
  ### Supervisor
  
  A sample Supervisor configuration is provided at `security_monkey_workers.conf`(https://github.com/Netflix/security_monkey/tree/develop/supervisor/security_monkey_workers.conf)
  that will automatically run the Celery worker processes for you.
  
  Run the following commands to set this up:
  
      sudo chgrp -R www-data /var/log/security_monkey
      sudo cp /usr/local/src/security_monkey/supervisor/security_monkey_workers.conf /etc/supervisor/conf.d/security_monkey_workers.conf
      sudo systemctl enable supervisor
      sudo systemctl start supervisor
      sudo supervisorctl status
  
  
  Supervisor will run the Celery `worker` command, which is:
  ```
   /usr/local/src/security_monkey/venv/bin/celery -A security_monkey.task_scheduler.tasks.CELERY worker
  ```
  
  **Note:** Please do not run this on the same instance as the scheduler. Exactly one scheduler should exist. There can be many worker instances,
  so keep the supervisor configurations on these instances separate.
  
  
  Deployment Strategies
  --------------------
  A typical deployment strategy is:
  1. Deploy the UI behind a Load Balancer
  1. Tear-down the scheduler instance
  1. Tear-down all worker instances
  1. *At this point, you should perform any database migrations and upgrades as there are no workers that will be affected.*
  1. Deploy the new scheduler instance -- wait for it to come online
  1. Deploy many worker instances
  
  Performing the steps in this order will ensure:
  - That no duplicate schedulers are running (multiple schedulers running causes havoc)
  - Proper DB upgrades occur without possibly impacting workers mutating the database
  - The schedulers and workers are working properly together
  
files: docs/autostarting.md
files: # encoding=utf8
  """
  .. module: security_monkey.task_scheduler.tasks
      :platform: Unix
      :synopsis: Sets up the Celery task scheduling for watching, auditing, and reporting changes in the environment.
  
  .. version:: $$VERSION$$
  .. moduleauthor:: Mike Grima <mgrima@netflix.com>
  
  """
  import sys
  
  try:
      reload(sys)  # Python 2
      sys.setdefaultencoding('utf8')
      # ^^ Fixes UTF-8 issues
  except NameError:
      pass  # Python 3
  
  import time
  import traceback
  
  from security_monkey import app, db, jirasync, sentry
  from security_monkey.alerter import Alerter
  from security_monkey.datastore import store_exception, clear_old_exceptions, Technology, Account, Item, ItemRevision
  from security_monkey.monitors import get_monitors, get_monitors_and_dependencies
  from security_monkey.reporter import Reporter
  from security_monkey.task_scheduler.util import CELERY, setup
  import boto3
  from sqlalchemy.exc import OperationalError, InvalidRequestError, StatementError
  
  
  @CELERY.task(bind=True, max_retries=3)
  def task_account_tech(self, account_name, technology_name):
      setup()
      app.logger.info("[ ] Executing Celery task for account: {}, technology: {}".format(account_name, technology_name))
      time1 = time.time()
  
      # Verify that the account exists (was it deleted? was it renamed?):
      if not Account.query.filter(Account.name == account_name).first():
          app.logger.error("[X] Account has been removed or renamed: {}. Please restart the scheduler to fix.".format(
              account_name
          ))
          return
  
      try:
          reporter_logic(account_name, technology_name)
  
          time2 = time.time()
          app.logger.info('[@] Run Account for Technology (%s/%s) took %0.1f s' % (account_name,
                                                                                   technology_name, (time2 - time1)))
          app.logger.info(
              "[+] Completed Celery task for account: {}, technology: {}".format(account_name, technology_name))
      except Exception as e:
          if sentry:
              sentry.captureException()
          app.logger.error("[X] Task Account Scheduler Exception ({}/{}): {}".format(account_name, technology_name, e))
          app.logger.error(traceback.format_exc())
          store_exception("scheduler-exception-on-watch", None, e)
          raise self.retry(exc=e)
  
  
  @CELERY.task(bind=True, max_retries=3)
  def task_audit(self, account_name, technology_name):
      setup()
  
      app.logger.info("[ ] Executing Celery task to audit changes for Account: {} Technology: {}".format(account_name,
                                                                                                         technology_name))
      # Verify that the account exists (was it deleted? was it renamed?):
      if not Account.query.filter(Account.name == account_name).first():
          app.logger.error("[X] Account has been removed or renamed: {}. Please restart the scheduler to fix.".format(
              account_name
          ))
          return
  
      try:
          audit_changes([account_name], [technology_name], True)
  
          app.logger.info("[+] Completed Celery task for account: {}, technology: {}".format(account_name,
                                                                                             technology_name))
  
      except Exception as e:
          if sentry:
              sentry.captureException()
          app.logger.error("[X] Task Audit Scheduler Exception ({}/{}): {}".format(account_name, technology_name, e))
          app.logger.error(traceback.format_exc())
          store_exception("scheduler-exception-on-audit", None, e)
          self.retry(exc=e)
  
  
  @CELERY.task()
  def clear_expired_exceptions():
      app.logger.info("[ ] Clearing out exceptions that have an expired TTL...")
      clear_old_exceptions()
      app.logger.info("[-] Completed clearing out exceptions that have an expired TTL.")
  
  
  def fix_orphaned_deletions(account_name, technology_name):
      """
      Possible issue with orphaned items. This will check if there are any, and will assume that the item
      was deleted. This will create a deletion change record to it.
  
      :param account_name:
      :param technology_name:
      :return:
      """
      # If technology doesn't exist, then create it:
      technology = Technology.query.filter(Technology.name == technology_name).first()
      if not technology:
          technology = Technology(name=technology_name)
          db.session.add(technology)
          db.session.commit()
          app.logger.info("Technology: {} did not exist... created it...".format(technology_name))
  
      account = Account.query.filter(Account.name == account_name).one()
  
      # Query for orphaned items of the given technology/account pair:
      orphaned_items = Item.query.filter(Item.account_id == account.id, Item.tech_id == technology.id,
                                         Item.latest_revision_id == None).all()  # noqa
  
      if not orphaned_items:
          app.logger.info("[@] No orphaned items have been found. (This is good)")
          return
  
      # Fix the orphaned items:
      for oi in orphaned_items:
          app.logger.error("[?] Found an orphaned item: {}. Creating a deletion record for it".format(oi.name))
          revision = ItemRevision(active=False, config={})
          oi.revisions.append(revision)
          db.session.add(revision)
          db.session.add(oi)
          db.session.commit()
  
          # Update the latest revision id:
          db.session.refresh(revision)
          oi.latest_revision_id = revision.id
          db.session.add(oi)
  
          db.session.commit()
          app.logger.info("[-] Created deletion record for item: {}.".format(oi.name))
  
  
  def reporter_logic(account_name, technology_name):
      """Logic for the run change reporter"""
      try:
          # Before doing anything... Look for orphaned items for this given technology. If they exist, then delete them:
          fix_orphaned_deletions(account_name, technology_name)
  
          # Watch and Audit:
          monitors = find_changes(account_name, technology_name)
  
          # Alert:
          app.logger.info("[ ] Sending alerts (if applicable) for account: {}, technology: {}".format(account_name,
                                                                                                      technology_name))
          Alerter(monitors, account=account_name).report()
      except (OperationalError, InvalidRequestError, StatementError) as e:
          app.logger.exception("[X] Database error processing account %s - technology %s cleaning up session.",
                               account_name, technology_name)
          db.session.remove()
          store_exception("scheduler-task-account-tech", None, e)
          raise e
  
  
  def manual_run_change_reporter(accounts):
      """Manual change reporting from the command line"""
      app.logger.info("[ ] Executing manual change reporter task...")
  
      try:
          for account in accounts:
              time1 = time.time()
              rep = Reporter(account=account)
  
              for monitor in rep.all_monitors:
                  if monitor.watcher:
                      app.logger.info("[ ] Running change finder for "
                                      "account: {} technology: {}".format(account, monitor.watcher.index))
                      reporter_logic(account, monitor.watcher.index)
  
              time2 = time.time()
              app.logger.info('[@] Run Account %s took %0.1f s' % (account, (time2 - time1)))
  
          app.logger.info("[+] Completed manual change reporting.")
      except (OperationalError, InvalidRequestError, StatementError) as e:
          app.logger.exception("[X] Database error processing cleaning up session.")
          db.session.remove()
          store_exception("scheduler-run-change-reporter", None, e)
          raise e
  
  
  def manual_run_change_finder(accounts, technologies):
      """Manual change finder"""
      app.logger.info("[ ] Executing manual find changes task...")
  
      try:
          for account in accounts:
              time1 = time.time()
  
              for tech in technologies:
                  find_changes(account, tech)
  
              time2 = time.time()
              app.logger.info('[@] Run Account %s took %0.1f s' % (account, (time2 - time1)))
          app.logger.info("[+] Completed manual change finder.")
      except (OperationalError, InvalidRequestError, StatementError) as e:
          app.logger.exception("[X] Database error processing cleaning up session.")
          db.session.remove()
          store_exception("scheduler-run-change-reporter", None, e)
          raise e
  
  
  def find_changes(account_name, monitor_name, debug=True):
      """
          Runs the watcher and stores the result, re-audits all types to account
          for downstream dependencies.
      """
      # Before doing anything... Look for orphaned items for this given technology. If they exist, then delete them:
      fix_orphaned_deletions(account_name, monitor_name)
  
      monitors = get_monitors(account_name, [monitor_name], debug)
  
      items = []
      for mon in monitors:
          cw = mon.watcher
          app.logger.info("[-->] Looking for changes in account: {}, technology: {}".format(account_name, cw.index))
          if mon.batch_support:
              batch_logic(mon, cw, account_name, debug)
          else:
              # Just fetch normally...
              (items, exception_map) = cw.slurp()
  
              _post_metric(
                  'queue_items_added',
                  len(items),
                  account_name=account_name,
                  tech=cw.i_am_singular
              )
  
              cw.find_changes(current=items, exception_map=exception_map)
  
              cw.save()
  
      # Batched monitors have already been monitored, and they will be skipped over.
      audit_changes([account_name], [monitor_name], False, debug, items_count=len(items))
      db.session.close()
  
      return monitors
  
  
  def audit_changes(accounts, monitor_names, send_report, debug=True, skip_batch=True, items_count=None):
      """
      Audits changes in the accounts
      :param accounts:
      :param monitor_names:
      :param send_report:
      :param debug:
      :param skip_batch:
      :return:
      """
      for account in accounts:
          monitors = get_monitors_and_dependencies(account, monitor_names, debug)
          for monitor in monitors:
              # Skip batch support monitors... They have already been monitored.
              if monitor.batch_support and skip_batch:
                  continue
  
              app.logger.debug("[-->] Auditing account: {}, technology: {}".format(account, monitor.watcher.index))
              _audit_changes(account, monitor.auditors, send_report, debug)
  
              _post_metric(
                  'queue_items_completed',
                  items_count,
                  account_name=account,
                  tech=monitor.watcher.i_am_singular
              )
  
  
  def batch_logic(monitor, current_watcher, account_name, debug):
      """
      Performs the batch watcher finding and auditing.
  
      TODO: Investigate how this could, in the future, be set to parallelize the batches.
      :param monitor:
      :param current_watcher:
      :param account_name:
      :param debug:
      :return:
      """
      # Fetch the full list of items that we need to obtain:
      _, exception_map = current_watcher.slurp_list()
      if len(exception_map) > 0:
          # Get the location tuple to collect the region:
          location = exception_map.keys()[0]
          if len(location) > 2:
              region = location[2]
          else:
              region = "unknown"
  
          exc_strings = [str(exc) for exc in exception_map.values()]
  
          app.logger.error("[X] Exceptions have caused nothing to be fetched for {technology}"
                           "/{account}/{region}..."
                           " CANNOT CONTINUE FOR THIS WATCHER!\n"
                           "Exceptions encountered were: {e}".format(technology=current_watcher.i_am_plural,
                                                                     account=account_name,
                                                                     region=region,
                                                                     e=",".join(exc_strings)))
          return
  
      while not current_watcher.done_slurping:
          app.logger.debug("[-->] Fetching a batch of {batch} items for {technology}/{account}.".format(
              batch=current_watcher.batched_size, technology=current_watcher.i_am_plural, account=account_name
          ))
          (items, exception_map) = current_watcher.slurp()
  
          _post_metric(
              'queue_items_added',
              len(items),
              account_name=account_name,
              tech=current_watcher.i_am_singular
          )
  
          audit_items = current_watcher.find_changes(current=items, exception_map=exception_map)
          _audit_specific_changes(monitor, audit_items, False, debug)
  
          _post_metric(
              'queue_items_completed',
              len(items),
              account_name=account_name,
              tech=current_watcher.i_am_singular
          )
  
      # Delete the items that no longer exist:
      app.logger.debug("[-->] Deleting all items for {technology}/{account} that no longer exist.".format(
          technology=current_watcher.i_am_plural, account=account_name
      ))
      current_watcher.find_deleted_batch(account_name)
  
  
  def _audit_changes(account, auditors, send_report, debug=True):
      """ Runs auditors on all items """
      try:
          for au in auditors:
              au.items = au.read_previous_items()
              au.audit_objects()
              au.save_issues()
              if send_report:
                  report = au.create_report()
                  au.email_report(report)
  
              if jirasync:
                  app.logger.info('[-->] Syncing {} issues on {} with Jira'.format(au.index, account))
                  jirasync.sync_issues([account], au.index)
      except (OperationalError, InvalidRequestError, StatementError) as e:
          app.logger.exception("[X] Database error processing accounts %s, cleaning up session.", account)
          db.session.remove()
          store_exception("scheduler-audit-changes", None, e)
  
  
  def _audit_specific_changes(monitor, audit_items, send_report, debug=True):
      """
      Runs the auditor on specific items that are passed in.
      :param monitor:
      :param audit_items:
      :param send_report:
      :param debug:
      :return:
      """
      try:
          for au in monitor.auditors:
              au.items = audit_items
              au.audit_objects()
              au.save_issues()
              if send_report:
                  report = au.create_report()
                  au.email_report(report)
  
              if jirasync:
                  app.logger.info('[-->] Syncing {} issues on {} with Jira'.format(au.index, monitor.watcher.accounts[0]))
                  jirasync.sync_issues(monitor.watcher.accounts, au.index)
      except (OperationalError, InvalidRequestError, StatementError) as e:
          app.logger.exception("[X] Database error processing accounts %s, cleaning up session.",
                               monitor.watcher.accounts[0])
          db.session.remove()
          store_exception("scheduler-audit-changes", None, e)
  
  
  def _post_metric(event_type, amount, account_name=None, tech=None):
      if not app.config.get('METRICS_ENABLED', False):
          return
  
      cw_client = boto3.client('cloudwatch', region_name=app.config.get('METRICS_POST_REGION', 'us-east-1'))
      cw_client.put_metric_data(
          Namespace=app.config.get('METRICS_NAMESPACE', 'securitymonkey'),
          MetricData=[
              {
                  'MetricName': event_type,
                  'Timestamp': int(time.time()),
                  'Value': amount,
                  'Unit': 'Count',
                  'Dimensions': [
                      {
                          'Name': 'tech',
                          'Value': tech
                      },
                      {
                          'Name': 'account_number',
                          'Value': Account.query.filter(Account.name == account_name).first().identifier
                      }
                  ]
              }
          ]
      )
  
files: security_monkey/task_scheduler/tasks.py
files: # encoding=utf8
  """
  .. module: security_monkey.task_scheduler.tasks
      :platform: Unix
      :synopsis: Sets up the Celery task scheduling for watching, auditing, and reporting changes in the environment.
  
  .. version:: $$VERSION$$
  .. moduleauthor:: Mike Grima <mgrima@netflix.com>
  
  """
  import sys
  
  try:
      reload(sys)  # Python 2
      sys.setdefaultencoding('utf8')
      # ^^ Fixes UTF-8 issues
  except NameError:
      pass  # Python 3
  
  import time
  import traceback
  
  from security_monkey import app, db, jirasync, sentry
  from security_monkey.alerter import Alerter
  from security_monkey.datastore import store_exception, clear_old_exceptions, Technology, Account, Item, ItemRevision
  from security_monkey.monitors import get_monitors, get_monitors_and_dependencies
  from security_monkey.reporter import Reporter
  from security_monkey.task_scheduler.util import CELERY, setup
  from sqlalchemy.exc import OperationalError, InvalidRequestError, StatementError
  
  
  @CELERY.task(bind=True, max_retries=3)
  def task_account_tech(self, account_name, technology_name):
      setup()
      app.logger.info("[ ] Executing Celery task for account: {}, technology: {}".format(account_name, technology_name))
      time1 = time.time()
  
      # Verify that the account exists (was it deleted? was it renamed?):
      if not Account.query.filter(Account.name == account_name).first():
          app.logger.error("[X] Account has been removed or renamed: {}. Please restart the scheduler to fix.".format(
              account_name
          ))
          return
  
      try:
          reporter_logic(account_name, technology_name)
  
          time2 = time.time()
          app.logger.info('[@] Run Account for Technology (%s/%s) took %0.1f s' % (account_name,
                                                                                   technology_name, (time2 - time1)))
          app.logger.info(
              "[+] Completed Celery task for account: {}, technology: {}".format(account_name, technology_name))
      except Exception as e:
          if sentry:
              sentry.captureException()
          app.logger.error("[X] Task Account Scheduler Exception ({}/{}): {}".format(account_name, technology_name, e))
          app.logger.error(traceback.format_exc())
          store_exception("scheduler-exception-on-watch", None, e)
          raise self.retry(exc=e)
  
  
  @CELERY.task(bind=True, max_retries=3)
  def task_audit(self, account_name, technology_name):
      setup()
  
      app.logger.info("[ ] Executing Celery task to audit changes for Account: {} Technology: {}".format(account_name,
                                                                                                         technology_name))
      # Verify that the account exists (was it deleted? was it renamed?):
      if not Account.query.filter(Account.name == account_name).first():
          app.logger.error("[X] Account has been removed or renamed: {}. Please restart the scheduler to fix.".format(
              account_name
          ))
          return
  
      try:
          audit_changes([account_name], [technology_name], True)
  
          app.logger.info("[+] Completed Celery task for account: {}, technology: {}".format(account_name,
                                                                                             technology_name))
  
      except Exception as e:
          if sentry:
              sentry.captureException()
          app.logger.error("[X] Task Audit Scheduler Exception ({}/{}): {}".format(account_name, technology_name, e))
          app.logger.error(traceback.format_exc())
          store_exception("scheduler-exception-on-audit", None, e)
          self.retry(exc=e)
  
  
  @CELERY.task()
  def clear_expired_exceptions():
      app.logger.info("[ ] Clearing out exceptions that have an expired TTL...")
      clear_old_exceptions()
      app.logger.info("[-] Completed clearing out exceptions that have an expired TTL.")
  
  
  def fix_orphaned_deletions(account_name, technology_name):
      """
      Possible issue with orphaned items. This will check if there are any, and will assume that the item
      was deleted. This will create a deletion change record to it.
  
      :param account_name:
      :param technology_name:
      :return:
      """
      # If technology doesn't exist, then create it:
      technology = Technology.query.filter(Technology.name == technology_name).first()
      if not technology:
          technology = Technology(name=technology_name)
          db.session.add(technology)
          db.session.commit()
          app.logger.info("Technology: {} did not exist... created it...".format(technology_name))
  
      account = Account.query.filter(Account.name == account_name).one()
  
      # Query for orphaned items of the given technology/account pair:
      orphaned_items = Item.query.filter(Item.account_id == account.id, Item.tech_id == technology.id,
                                         Item.latest_revision_id == None).all()  # noqa
  
      if not orphaned_items:
          app.logger.info("[@] No orphaned items have been found. (This is good)")
          return
  
      # Fix the orphaned items:
      for oi in orphaned_items:
          app.logger.error("[?] Found an orphaned item: {}. Creating a deletion record for it".format(oi.name))
          revision = ItemRevision(active=False, config={})
          oi.revisions.append(revision)
          db.session.add(revision)
          db.session.add(oi)
          db.session.commit()
  
          # Update the latest revision id:
          db.session.refresh(revision)
          oi.latest_revision_id = revision.id
          db.session.add(oi)
  
          db.session.commit()
          app.logger.info("[-] Created deletion record for item: {}.".format(oi.name))
  
  
  def reporter_logic(account_name, technology_name):
      """Logic for the run change reporter"""
      try:
          # Before doing anything... Look for orphaned items for this given technology. If they exist, then delete them:
          fix_orphaned_deletions(account_name, technology_name)
  
          # Watch and Audit:
          monitors = find_changes(account_name, technology_name)
  
          # Alert:
          app.logger.info("[ ] Sending alerts (if applicable) for account: {}, technology: {}".format(account_name,
                                                                                                      technology_name))
          Alerter(monitors, account=account_name).report()
      except (OperationalError, InvalidRequestError, StatementError) as e:
          app.logger.exception("[X] Database error processing account %s - technology %s cleaning up session.",
                               account_name, technology_name)
          db.session.remove()
          store_exception("scheduler-task-account-tech", None, e)
          raise e
  
  
  def manual_run_change_reporter(accounts):
      """Manual change reporting from the command line"""
      app.logger.info("[ ] Executing manual change reporter task...")
  
      try:
          for account in accounts:
              time1 = time.time()
              rep = Reporter(account=account)
  
              for monitor in rep.all_monitors:
                  if monitor.watcher:
                      app.logger.info("[ ] Running change finder for "
                                      "account: {} technology: {}".format(account, monitor.watcher.index))
                      reporter_logic(account, monitor.watcher.index)
  
              time2 = time.time()
              app.logger.info('[@] Run Account %s took %0.1f s' % (account, (time2 - time1)))
  
          app.logger.info("[+] Completed manual change reporting.")
      except (OperationalError, InvalidRequestError, StatementError) as e:
          app.logger.exception("[X] Database error processing cleaning up session.")
          db.session.remove()
          store_exception("scheduler-run-change-reporter", None, e)
          raise e
  
  
  def manual_run_change_finder(accounts, technologies):
      """Manual change finder"""
      app.logger.info("[ ] Executing manual find changes task...")
  
      try:
          for account in accounts:
              time1 = time.time()
  
              for tech in technologies:
                  find_changes(account, tech)
  
              time2 = time.time()
              app.logger.info('[@] Run Account %s took %0.1f s' % (account, (time2 - time1)))
          app.logger.info("[+] Completed manual change finder.")
      except (OperationalError, InvalidRequestError, StatementError) as e:
          app.logger.exception("[X] Database error processing cleaning up session.")
          db.session.remove()
          store_exception("scheduler-run-change-reporter", None, e)
          raise e
  
  
  def find_changes(account_name, monitor_name, debug=True):
      """
          Runs the watcher and stores the result, re-audits all types to account
          for downstream dependencies.
      """
      # Before doing anything... Look for orphaned items for this given technology. If they exist, then delete them:
      fix_orphaned_deletions(account_name, monitor_name)
  
      monitors = get_monitors(account_name, [monitor_name], debug)
      for mon in monitors:
          cw = mon.watcher
          app.logger.info("[-->] Looking for changes in account: {}, technology: {}".format(account_name, cw.index))
          if mon.batch_support:
              batch_logic(mon, cw, account_name, debug)
          else:
              # Just fetch normally...
              (items, exception_map) = cw.slurp()
              cw.find_changes(current=items, exception_map=exception_map)
              cw.save()
  
      # Batched monitors have already been monitored, and they will be skipped over.
      audit_changes([account_name], [monitor_name], False, debug)
      db.session.close()
  
      return monitors
  
  
  def audit_changes(accounts, monitor_names, send_report, debug=True, skip_batch=True):
      """
      Audits changes in the accounts
      :param accounts:
      :param monitor_names:
      :param send_report:
      :param debug:
      :param skip_batch:
      :return:
      """
      for account in accounts:
          monitors = get_monitors_and_dependencies(account, monitor_names, debug)
          for monitor in monitors:
              # Skip batch support monitors... They have already been monitored.
              if monitor.batch_support and skip_batch:
                  continue
  
              app.logger.debug("[-->] Auditing account: {}, technology: {}".format(account, monitor.watcher.index))
              _audit_changes(account, monitor.auditors, send_report, debug)
  
  
  def batch_logic(monitor, current_watcher, account_name, debug):
      """
      Performs the batch watcher finding and auditing.
  
      TODO: Investigate how this could, in the future, be set to parallelize the batches.
      :param monitor:
      :param current_watcher:
      :param account_name:
      :param debug:
      :return:
      """
      # Fetch the full list of items that we need to obtain:
      _, exception_map = current_watcher.slurp_list()
      if len(exception_map) > 0:
          # Get the location tuple to collect the region:
          location = exception_map.keys()[0]
          if len(location) > 2:
              region = location[2]
          else:
              region = "unknown"
  
          exc_strings = [str(exc) for exc in exception_map.values()]
  
          app.logger.error("[X] Exceptions have caused nothing to be fetched for {technology}"
                           "/{account}/{region}..."
                           " CANNOT CONTINUE FOR THIS WATCHER!\n"
                           "Exceptions encountered were: {e}".format(technology=current_watcher.i_am_plural,
                                                                     account=account_name,
                                                                     region=region,
                                                                     e=",".join(exc_strings)))
          return
  
      while not current_watcher.done_slurping:
          app.logger.debug("[-->] Fetching a batch of {batch} items for {technology}/{account}.".format(
              batch=current_watcher.batched_size, technology=current_watcher.i_am_plural, account=account_name
          ))
          (items, exception_map) = current_watcher.slurp()
  
          audit_items = current_watcher.find_changes(current=items, exception_map=exception_map)
          _audit_specific_changes(monitor, audit_items, False, debug)
  
      # Delete the items that no longer exist:
      app.logger.debug("[-->] Deleting all items for {technology}/{account} that no longer exist.".format(
          technology=current_watcher.i_am_plural, account=account_name
      ))
      current_watcher.find_deleted_batch(account_name)
  
  
  def _audit_changes(account, auditors, send_report, debug=True):
      """ Runs auditors on all items """
      try:
          for au in auditors:
              au.items = au.read_previous_items()
              au.audit_objects()
              au.save_issues()
              if send_report:
                  report = au.create_report()
                  au.email_report(report)
  
              if jirasync:
                  app.logger.info('[-->] Syncing {} issues on {} with Jira'.format(au.index, account))
                  jirasync.sync_issues([account], au.index)
      except (OperationalError, InvalidRequestError, StatementError) as e:
          app.logger.exception("[X] Database error processing accounts %s, cleaning up session.", account)
          db.session.remove()
          store_exception("scheduler-audit-changes", None, e)
  
  
  def _audit_specific_changes(monitor, audit_items, send_report, debug=True):
      """
      Runs the auditor on specific items that are passed in.
      :param monitor:
      :param audit_items:
      :param send_report:
      :param debug:
      :return:
      """
      try:
          for au in monitor.auditors:
              au.items = audit_items
              au.audit_objects()
              au.save_issues()
              if send_report:
                  report = au.create_report()
                  au.email_report(report)
  
              if jirasync:
                  app.logger.info('[-->] Syncing {} issues on {} with Jira'.format(au.index, monitor.watcher.accounts[0]))
                  jirasync.sync_issues(monitor.watcher.accounts, au.index)
      except (OperationalError, InvalidRequestError, StatementError) as e:
          app.logger.exception("[X] Database error processing accounts %s, cleaning up session.",
                               monitor.watcher.accounts[0])
          db.session.remove()
          store_exception("scheduler-audit-changes", None, e)
  
files: security_monkey/task_scheduler/tasks.py