# Part of Odoo. See LICENSE file for full copyright and licensing details.
importodoo
importsys
importoptparse
importlogging
fromcollectionsimportdefaultdict
from.importCommand
fromodoo.modules.registryimportRegistry
fromodoo.toolsimportSQL
_logger=logging.getLogger(__name__)
classObfuscate(Command):
"""Obfuscate data in a given odoo database"""
def__init__(self):
super().__init__()
self.cr=None
def_ensure_cr(func):
defcheck_cr(self,*args,**kwargs):
ifnotself.cr:
raiseException("No database connection")
returnfunc(self,*args,**kwargs)
returncheck_cr
@_ensure_cr
defbegin(self):
self.cr.execute("begin work")
self.cr.execute("CREATE EXTENSION IF NOT EXISTS pgcrypto")
@_ensure_cr
defcommit(self):
self.cr.commit()
@_ensure_cr
defrollback(self):
self.cr.rollback()
@_ensure_cr
defset_pwd(self,pwd):
"""Set password to cypher/uncypher datas"""
self.cr.execute("INSERT INTO ir_config_parameter (key, value) VALUES ('odoo_cyph_pwd', 'odoo_cyph_'||encode(pgp_sym_encrypt(%s, %s), 'base64')) ON CONFLICT(key) DO NOTHING",[pwd,pwd])
self.cr.execute("DELETE FROM ir_config_parameter WHERE key='odoo_cyph_pwd'")
defcypher_string(self,sql_field:SQL,password):
# don't double cypher fields
returnSQL("""CASE WHEN starts_with(%(field_name)s, 'odoo_cyph_') THEN %(field_name)s ELSE 'odoo_cyph_'||encode(pgp_sym_encrypt(%(field_name)s, %(pwd)s), 'base64') END""",field_name=sql_field,pwd=password)
defuncypher_string(self,sql_field:SQL,password):
returnSQL("""CASE WHEN starts_with(%(field_name)s, 'odoo_cyph_') THEN pgp_sym_decrypt(decode(substring(%(field_name)s, 11)::text, 'base64'), %(pwd)s) ELSE %(field_name)s END""",field_name=sql_field,pwd=password)
defcheck_field(self,table,field):
qry="SELECT udt_name FROM information_schema.columns WHERE table_name=%s AND column_name=%s"
self.cr.execute(qry,[table,field])
ifself.cr.rowcount==1:
res=self.cr.fetchone()
ifres[0]in['text','varchar']:
# Doesn t work for selection fields ...
return'string'
ifres[0]=='jsonb':
return'json'
returnFalse
defget_all_fields(self):
qry="SELECT table_name, column_name FROM information_schema.columns WHERE table_schema='public' AND udt_name IN ['text', 'varchar', 'jsonb'] AND NOT table_name LIKE 'ir_%' ORDER BY 1,2"
group.add_option('--fields',dest="fields",default=False,help="List of table.columns to obfuscate/unobfuscate: table1.column1,table2.column1,table2.column2")
group.add_option('--exclude',dest="exclude",default=False,help="List of table.columns to exclude from obfuscate/unobfuscate: table1.column1,table2.column1,table2.column2")
group.add_option('--file',dest="file",default=False,help="File containing the list of table.columns to obfuscate/unobfuscate")
group.add_option('--allfields',action='store_true',default=False,help="Used in unobfuscate mode, try to unobfuscate all fields. Cannot be used in obfuscate mode. Slower than specifying fields.")
group.add_option('--vacuum',action='store_true',default=False,help="Vacuum database after unobfuscating")
group.add_option('--pertablecommit',action='store_true',default=False,help="Commit after each table instead of a big transaction")
help="Don't ask for manual confirmation. Use it carefully as the obfuscate method is not considered as safe to transfer anonymous datas to a third party.")