#!/usr/bin/env python
#
# Public Domain 2014-present MongoDB, Inc.
# Public Domain 2008-2014 WiredTiger, Inc.
#
# This is free and unencumbered software released into the public domain.
#
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled
# binary, for any purpose, commercial or non-commercial, and by any
# means.
#
# In jurisdictions that recognize copyright laws, the author or authors
# of this software dedicate any and all copyright interest in the
# software to the public domain. We make this dedication for the benefit
# of the public at large and to the detriment of our heirs and
# successors. We intend this dedication to be an overt act of
# relinquishment in perpetuity of all present and future rights to this
# software under copyright law.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
#
# test_encrypt06.py
#   Test that encryption is effective, it leaves no clear text
#

import os, run, random
import wiredtiger, wttest
from wtscenario import make_scenarios

# Test encryption, when on, does not leak any information
class test_encrypt06(wttest.WiredTigerTestCase):

    key11 = ',keyid=11,secretkey=XYZ'
    key13 = ',keyid=13'

    # Test with various combinations of tables with or without indices
    # and column groups, also with LSM.  When 'match' is False, we
    # testing a potential misuse of the API: a table is opened with
    # with its own encryption options (different from the system),
    # but the indices and column groups do not specify encryption,
    # so they may get the system encryptor.
    storagetype = [
        ('table', dict(
            uriprefix='table:', use_cg=False, use_index=False, match=True)),
        ('table-idx', dict(
            uriprefix='table:', use_cg=False, use_index=True, match=True)),
        ('table-cg', dict(
            uriprefix='table:', use_cg=True, use_index=False, match=True)),
        ('table-cg-idx', dict(
            uriprefix='table:', use_cg=True, use_index=True, match=True)),
        ('table-idx-unmatch', dict(
            uriprefix='table:', use_cg=False, use_index=True, match=False)),
        ('table-cg-unmatch', dict(
            uriprefix='table:', use_cg=True, use_index=False, match=False)),
        ('table-cg-idx-unmatch', dict(
            uriprefix='table:', use_cg=True, use_index=True, match=False)),
        ('lsm', dict(
            uriprefix='lsm:', use_cg=False, use_index=False, match=True)),
    ]
    encrypt = [
        ('none', dict(
            sys_encrypt='none', sys_encrypt_args='',
            table0_encrypt='none', table0_encrypt_args='',
            table1_encrypt='none', table1_encrypt_args='')),
        ('rotn-implied', dict(
            sys_encrypt='rotn', sys_encrypt_args=key11,
            table0_encrypt=None, table0_encrypt_args='',
            table1_encrypt=None, table1_encrypt_args='')),
        ('rotn-all', dict(
            sys_encrypt='rotn', sys_encrypt_args=key11,
            table0_encrypt='rotn', table0_encrypt_args=key13,
            table1_encrypt='rotn', table1_encrypt_args=key13)),
        ('rotn-sys', dict(
            sys_encrypt='rotn', sys_encrypt_args=key11,
            table0_encrypt='none', table0_encrypt_args='',
            table1_encrypt='none', table1_encrypt_args='')),
        ('rotn-table0', dict(
            sys_encrypt='rotn', sys_encrypt_args=key11,
            table0_encrypt='rotn', table0_encrypt_args=key13,
            table1_encrypt='none', table1_encrypt_args='')),
    ]
    scenarios = make_scenarios(encrypt, storagetype)
    nrecords = 1000

    def conn_extensions(self, extlist):
        extlist.skip_if_missing = True
        extlist.extension('encryptors', self.sys_encrypt)
        extlist.extension('encryptors', self.table0_encrypt)
        extlist.extension('encryptors', self.table1_encrypt)

    def conn_config(self):
        return 'encryption=(name={0}{1}),'.format(
            self.sys_encrypt, self.sys_encrypt_args)

    def encrypt_table_params(self, name, args):
        if name == None:
            return ''
        else:
            return ',encryption=(name=' + name + args + ')'

    def match_string_in_file(self, fname, match):
        with open(fname, 'rb') as f:
            return (f.read().find(match) != -1)

    def match_string_in_rundir(self, match):
        byte_match = match.encode()
        for fname in os.listdir('.'):
            if self.match_string_in_file(fname, byte_match):
                return True
        return False

    def visible_data(self, table_setting):
        if table_setting == None:
            # No table encryption explicitly set, so we use the system setting
            visible = (self.sys_encrypt == 'none')
        else:
            visible = (table_setting == 'none')
        return visible

    def visible_name(self, table_setting, iskey):
        if table_setting == None:
            # No table encryption explicitly set, so we use the system setting
            visible = (self.sys_encrypt == 'none')
        else:
            visible = (table_setting == 'none')

        # If we have everything in a column group, the key name will not
        # be stored in the column group files.  It will be stored in the
        # system metadata, if that is not encrypted.
        if iskey and self.use_cg and self.sys_encrypt != 'none':
            visible = False
        return visible

    # Create a table, add key/values with specific lengths, then verify them.
    def test_encrypt(self):
        name0 = 'test_encrypt06-0'
        name1 = 'test_encrypt06-1'

        enc0 = self.encrypt_table_params(self.table0_encrypt,
                                        self.table0_encrypt_args)
        enc1 = self.encrypt_table_params(self.table1_encrypt,
                                        self.table1_encrypt_args)

        # This is the clear text that we'll be looking for
        txt0 = 'AbCdEfG'
        txt1 = 'aBcDeFg'
        keyname0 = 'MyKey0Name'
        keyname1 = 'MyKey1Name'
        valname0 = 'MyValue0Name'
        valname1 = 'MyValue1Name'

        # Make a bunch of column group and indices,
        # we want to see if any information is leaked anywhere.
        # The key column and one of the value columns is given a name
        # we will look for as clear text.
        sharedparam = 'key_format=S,value_format=SSSS,' + \
                      'columns=({},{},v1,v2,v3),'

        s = self.session
        pfx = self.uriprefix

        cgparam = 'colgroups=(g00,g01)' if self.use_cg else ''
        s.create(pfx + name0, sharedparam.format(keyname0, valname0) + \
                 cgparam + enc0)

        if not self.match:
            enc0 = ''
        if self.use_cg:
            s.create('colgroup:' + name0 + ':g00',
                     'columns=({},v1)'.format(valname0) + enc0)
            s.create('colgroup:' + name0 + ':g01', 'columns=(v2,v3)' + enc0)
        if self.use_index:
            s.create('index:' + name0 + ':i00',
                     'columns=({})'.format(valname0) + enc0)
            s.create('index:' + name0 + ':i01', 'columns=(v1,v2)' + enc0)
            s.create('index:' + name0 + ':i02', 'columns=(v3)' + enc0)

        cgparam = 'colgroups=(g10,g11)' if self.use_cg else ''
        s.create(pfx + name1, sharedparam.format(keyname1, valname1) + \
                 cgparam + enc1)

        if not self.match:
            enc1 = ''
        if self.use_cg:
            s.create('colgroup:' + name1 + ':g10',
                     'columns=({},v1)'.format(valname1) + enc1)
            s.create('colgroup:' + name1 + ':g11', 'columns=(v2,v3)' + enc1)
        if self.use_index:
            s.create('index:' + name1 + ':i10',
                     'columns=({})'.format(valname1) + enc1)
            s.create('index:' + name1 + ':i11', 'columns=(v1,v2)' + enc1)
            s.create('index:' + name1 + ':i12', 'columns=(v3)' + enc1)

        c0 = s.open_cursor(pfx + name0, None)
        c1 = s.open_cursor(pfx + name1, None)
        for idx in range(1,self.nrecords):
            c0.set_key(str(idx) + txt0)
            c1.set_key(str(idx) + txt1)
            c0.set_value(txt0 * (idx % 97), txt0 * 3, txt0 * 5, txt0 * 7)
            c1.set_value(txt1 * (idx % 97), txt1 * 3, txt1 * 5, txt1 * 7)
            c0.insert()
            c1.insert()

        c0.close()
        c1.close()

        # Force everything to disk so we can examine it
        self.close_conn()

        if self.match:
            # Key and value names are encrypted according to the
            # encryption level on the associated table.
            self.assertEqual(self.visible_data(self.table0_encrypt),
                self.match_string_in_rundir(txt0))
            self.assertEqual(self.visible_name(self.table0_encrypt, True),
                self.match_string_in_rundir(keyname0))
            self.assertEqual(self.visible_name(self.table0_encrypt, False),
                self.match_string_in_rundir(valname0))

            self.assertEqual(self.visible_data(self.table1_encrypt),
                self.match_string_in_rundir(txt1))
            self.assertEqual(self.visible_name(self.table1_encrypt, True),
                self.match_string_in_rundir(keyname1))
            self.assertEqual(self.visible_name(self.table1_encrypt, False),
                self.match_string_in_rundir(valname1))
        else:
            # If the encryption config for indices and column groups is blank,
            # we make a conservative check - if we specified encryption on the
            # table, none of our data or key/value names should be exposed.
            #
            # If we have system encryption on, set table encryption to 'none',
            # and set the index or column group config to blank, we technically
            # should get no encryption for names or data. That currently
            # doesn't work (CGs and indices instead will be encrypted),
            # so we don't cover that case.
            if self.table0_encrypt != 'none':
                self.assertFalse(self.match_string_in_rundir(txt0))
                self.assertFalse(self.match_string_in_rundir(keyname0))
                self.assertFalse(self.match_string_in_rundir(valname0))
            if self.table1_encrypt != 'none':
                self.assertFalse(self.match_string_in_rundir(txt1))
                self.assertFalse(self.match_string_in_rundir(keyname1))
                self.assertFalse(self.match_string_in_rundir(valname1))

if __name__ == '__main__':
    wttest.run()
