# Copyright 2016 Capital One Services, LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""S3 Resource Manager

Filters:

The generic Values filters (jmespath) expression and Or filter are
available with all resources, including buckets, we include several
additonal bucket data (Tags, Replication, Acl, Policy) as keys within
a bucket representation.

Actions:

 encrypt-keys

   Scan all keys in a bucket and optionally encrypt them in place.

 global-grants

   Check bucket acls for global grants

 encryption-policy

   Attach an encryption required policy to a bucket, this will break
   applications that are not using encryption, including aws log
   delivery.

"""
from botocore.client import Config
from botocore.exceptions import ClientError
from botocore.vendored.requests.exceptions import SSLError
from concurrent.futures import as_completed

import functools
import json
import itertools
import logging
import math
import os
import time
import ssl

from c7n import executor
from c7n.actions import ActionRegistry, BaseAction, AutoTagUser
from c7n.filters import FilterRegistry, Filter, CrossAccountAccessFilter
from c7n.manager import resources
from c7n.query import QueryResourceManager, ResourceQuery
from c7n.tags import Tag
from c7n.utils import chunks, local_session, set_annotation, type_schema

"""
TODO:
 - How does replication status effect in place encryption.
 - Test glacier support
"""

log = logging.getLogger('custodian.s3')

filters = FilterRegistry('s3.filters')
actions = ActionRegistry('s3.actions')
actions.register('auto-tag-user', AutoTagUser)

MAX_COPY_SIZE = 1024 * 1024 * 1024 * 2


@resources.register('s3')
class S3(QueryResourceManager):

    class resource_type(ResourceQuery.resolve("aws.s3.bucket")):
        dimension = 'BucketName'

    executor_factory = executor.ThreadPoolExecutor
    filter_registry = filters
    action_registry = actions

    def __init__(self, ctx, data):
        super(S3, self).__init__(ctx, data)
        self.log_dir = ctx.log_dir

    def augment(self, buckets):
        with self.executor_factory(
                max_workers=min((10, len(buckets)))) as w:
            results = w.map(
                assemble_bucket,
                zip(itertools.repeat(self.session_factory), buckets))
            results = filter(None, results)
            return results


S3_AUGMENT_TABLE = (
    ('get_bucket_location', 'Location', None, None),
    ('get_bucket_tagging', 'Tags', [], 'TagSet'),
    ('get_bucket_policy',  'Policy', None, 'Policy'),
    ('get_bucket_acl', 'Acl', None, None),
    ('get_bucket_replication', 'Replication', None, None),
    ('get_bucket_versioning', 'Versioning', None, None),
    ('get_bucket_website', 'Website', None, None),
    ('get_bucket_logging', 'Logging', None, 'LoggingEnabled'),
    ('get_bucket_notification_configuration', 'Notification', None, None)
#        ('get_bucket_lifecycle', 'Lifecycle', None, None),
#        ('get_bucket_cors', 'Cors'),
)


def assemble_bucket(item):
    """Assemble a document representing all the config state around a bucket.
    """
    factory, b = item
    s = factory()
    c = s.client('s3')
    # Bucket Location, Current Client Location, Default Location
    b_location = c_location = location = "us-east-1"
    methods = list(S3_AUGMENT_TABLE)
    for m, k, default, select in methods:
        try:
            method = getattr(c, m)
            v = method(Bucket=b['Name'])
            v.pop('ResponseMetadata')
            if select is not None and select in v:
                v = v[select]
        except (ssl.SSLError, SSLError) as e:
            # Proxy issues? i assume
            log.warning("Bucket ssl error %s: %s %s",
                        b['Name'], b.get('Location', 'unknown'),
                        e)
            continue
        except ClientError as e:
            code =  e.response['Error']['Code']
            if code.startswith("NoSuch") or "NotFound" in code:
                v = default
            elif code == 'PermanentRedirect':
                s = factory()
                c = bucket_client(s, b)
                # Requeue with the correct region given location constraint
                methods.append((m, k, default, select))
                continue
            else:
                log.warning(
                    "Bucket:%s unable to invoke method:%s error:%s " % (
                        b['Name'], m, e.response['Error']['Message']))
                return None
        # As soon as we learn location (which generally works)
        if k == 'Location' and v is not None:
            b_location = v.get('LocationConstraint')
            if v and v != c_location:
                c = s.client('s3', region_name=b_location)
            elif c_location != location:
                c = s.client('s3', region_name=location)
        b[k] = v
    return b


def bucket_client(session, b, kms=False):
    location = b.get('Location')
    if location is None:
        region = 'us-east-1'
    else:
        region = location['LocationConstraint'] or 'us-east-1'
    if kms:
        # Need v4 signature for aws:kms crypto
        config = Config(signature_version='s3v4', read_timeout=200)
    else:
        config = Config(read_timeout=200)
    return session.client('s3', region_name=region, config=config)


@filters.register('cross-account')
class S3CrossAccountFilter(CrossAccountAccessFilter):

    def get_accounts(self):
        """add in elb access by default

        ELB Accounts by region http://goo.gl/a8MXxd
        """
        accounts = super(S3CrossAccountFilter, self).get_accounts()
        return accounts.union(
            ['127311923021',  # us-east-1
             '797873946194',  # us-west-2
             '027434742980',  # us-west-1
             '156460612806',  # eu-west-1
             '054676820928',  # eu-central-1
             '114774131450',  # ap-southeast-1
             '582318560864',  # ap-northeast-1
             '783225319266',  # ap-southeast-2
             '600734575887',  # ap-northeast-2
             '507241528517',  # sa-east-1
             '048591011584',  # gov-cloud-1
             ])


@filters.register('global-grants')
class GlobalGrantsFilter(Filter):

    schema = type_schema('global-grants', permissions={
        'type': 'array', 'items': {
            'type': 'string', 'enum': [
                'READ', 'WRITE', 'WRITE_ACP', 'READ', 'READ_ACP']}})

    GLOBAL_ALL = "http://acs.amazonaws.com/groups/global/AllUsers"
    AUTH_ALL = "http://acs.amazonaws.com/groups/global/AuthenticatedUsers"

    def process(self, buckets, event=None):
        with self.executor_factory(max_workers=5) as w:
            results = w.map(self.process_bucket, buckets)
            results = filter(None, list(results))
            return results

    def process_bucket(self, b):
        acl = b.get('Acl', {'Grants': []})
        if not acl or not acl['Grants']:
            return
        results = []
        perms = self.data.get('permissions', [])
        for grant in acl['Grants']:
            if 'URI' not in grant.get("Grantee", {}):
                continue
            if grant['Grantee']['URI'] not in [self.AUTH_ALL, self.GLOBAL_ALL]:
                continue
            if grant['Permission'] == 'READ' and b['Website']:
                continue
            if not perms or (perms and grant['Permission'] in perms):
                results.append(grant['Permission'])

        c = bucket_client(self.manager.session_factory(), b)

        if results:
            set_annotation(b, 'GlobalPermissions', results)
            return b


class BucketActionBase(BaseAction):

    def get_permissions(self):
        return self.permissions


@filters.register('has-statement')
class HasStatementFilter(Filter):
    """Find buckets with set of named policy statements."""
    schema = type_schema(
        'has-statement',
        statement_ids={'type': 'array', 'items': {'type': 'string'}})

    def process(self, buckets, event=None):
        return filter(None, map(self.process_bucket, buckets))

    def process_bucket(self, b):
        p = b.get('Policy')
        if p is None:
            return b
        p = json.loads(p)
        required = list(self.data.get('statement_ids', []))
        statements = p.get('Statement', [])
        for s in list(statements):
            if s.get('Sid') in required:
                required.remove(s['Sid'])
        if not required:
            return b
        return None


@filters.register('missing-statement')
@filters.register('missing-policy-statement')
class MissingPolicyStatementFilter(Filter):
    """Find buckets missing a set of named policy statements."""

    schema = type_schema(
        'missing-policy-statement',
        aliases=('missing-statement',),
        statement_ids={'type': 'array', 'items': {'type': 'string'}})

    def __call__(self, b):
        p = b.get('Policy')
        if p is None:
            return b

        p = json.loads(p)

        required = list(self.data.get('statement_ids', []))
        statements = p.get('Statement', [])
        for s in list(statements):
            if s.get('Sid') in required:
                required.remove(s['Sid'])
        if not required:
            return False
        return True


@actions.register('no-op')
class NoOp(BucketActionBase):

    schema = type_schema('no-op')

    def process(self, buckets):
        return None


@actions.register('remove-statements')
class RemovePolicyStatement(BucketActionBase):

    schema = type_schema(
        'remove-statements',
        statement_ids={'type': 'array', 'items': {'type': 'string'}})

    def process(self, buckets):
        with self.executor_factory(max_workers=3) as w:
            results = w.map(self.process_bucket, buckets)
            return filter(None, list(results))

    def process_bucket(self, bucket):
        p = bucket.get('Policy')
        if p is None:
            return
        else:
            p = json.loads(p)

        statements = p.get('Statement', [])
        found = []
        for s in list(statements):
            if s['Sid'] in self.data['statement_ids']:
                found.append(s)
                statements.remove(s)
        if not found:
            return

        s3 = local_session(self.manager.session_factory).client('s3')
        if not statements:
            s3.delete_bucket_policy(Bucket=bucket['Name'])
        else:
            s3.put_bucket_policy(Bucket=bucket['Name'], Policy=json.dumps(p))
        return {'Name': bucket['Name'], 'State': 'PolicyRemoved', 'Statements': found}


@actions.register('attach-encrypt')
class AttachLambdaEncrypt(BucketActionBase):
    schema = type_schema(
        'attach-encrypt', role={'type': 'string'})

    def __init__(self, data=None, manager=None):
        self.data = data or {}
        self.manager = manager

    def validate(self):
        if not self.data.get('role', self.manager.config.assume_role):
            raise ValueError(
                "attach-encrypt: role must be specified either"
                "via assume or in config")