-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathdbutils.py
602 lines (564 loc) · 24.2 KB
/
dbutils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
import cx_Oracle
from tenacity import retry
import tenacity
import logging
import envvars
import yaml
import os
import datetime as dt
import uuid
import re
STATUS_OK = 'ok'
STATUS_ERROR = 'error'
logger = logging.getLogger(__name__)
class dbConfig(object):
def __init__(self, all_databases):
file = os.path.join(
os.path.dirname(__file__),
"oracle_user_manager.yaml"
)
with open(file, 'r') as cfile:
self.conf = yaml.load(cfile, Loader=yaml.FullLoader)
self.databases = all_databases
@retry(reraise=True, stop=tenacity.stop.stop_after_attempt(2), wait=tenacity.wait.wait_fixed(2))
def __login(self, user, passwd, dsn):
logger.info('Connecting to DB [{}] as {}...'.format(dsn, user))
try:
dbh = cx_Oracle.connect(user, passwd, dsn=dsn)
dbh.close()
return True, "", False
except Exception as e:
raise e
def is_alphanumeric(self, in_string):
return re.fullmatch(r'^[A-Za-z0-9]+$', in_string)
def get_username_from_email(self, email, db=None):
if not db:
db = self.databases[0]
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['user'], self.conf[db]['passwd'], dsn=dsn)
cursor = dbh.cursor()
username = None
try:
sql = """
SELECT USERNAME, EMAIL from DES_ADMIN.DES_USERS where EMAIL = :email
"""
for row in cursor.execute(sql, email=email.lower()):
username, email = row
except Exception as e:
logger.error(str(e).strip())
cursor.close()
dbh.close()
return username
def check_credentials(self, username, password, db, email=''):
if not db:
db = self.databases[0]
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
update = False
try:
# Get username from account with registered email address if provided
if email != '':
username_from_email = self.get_username_from_email(email, db)
if not username_from_email:
return False, username, 'email is not registered', update
else:
username = username_from_email
# If email is not provided, require username parameter
if not username:
return False, username, 'Registered username or email is required', update
auth, error, update = self.__login(username, password, dsn)
return auth, username, error, update
except Exception as e:
error = str(e).strip()
if '28001' in error:
update = True
return False, username, error, update
def get_basic_info(self, user, db=None):
if not db:
db = self.databases[0]
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['user'], self.conf[db]['passwd'], dsn=dsn)
cursor = dbh.cursor()
try:
if user == envvars.MONITOR_SERVICE_ACCOUNT_USERNAME:
cc = (envvars.MONITOR_SERVICE_ACCOUNT_USERNAME, envvars.MONITOR_SERVICE_ACCOUNT_USERNAME, '[email protected]')
else:
sql = """
SELECT FIRSTNAME, LASTNAME, EMAIL from DES_ADMIN.DES_USERS WHERE USERNAME = :username
"""
cc = cursor.execute(sql, username=user).fetchone()
except:
cc = ('','','')
cursor.close()
dbh.close()
return cc
def list_all_users(self, db=None):
if not db:
db = self.databases[0]
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['user'], self.conf[db]['passwd'], dsn=dsn)
cursor = dbh.cursor()
try:
sql = """
SELECT USERNAME, FIRSTNAME, LASTNAME, EMAIL from DES_ADMIN.DES_USERS
"""
cc = cursor.execute(sql).fetchall()
except:
cc = ('','','','')
cursor.close()
dbh.close()
return cc
def update_info(self, username, firstname, lastname, email):
for db in self.databases:
username = username.lower()
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['user'], self.conf[db]['passwd'], dsn=dsn)
cursor = dbh.cursor()
qupdate = """
UPDATE DES_ADMIN.DES_USERS SET
FIRSTNAME = :first,
LASTNAME = :last,
EMAIL = :email
WHERE USERNAME = :username
"""
try:
cursor.execute(qupdate, first=firstname, last=lastname, email=email, username=username)
dbh.commit()
msg = 'Information for {} Updated'.format(username)
status = 'ok'
except Exception as e:
msg = str(e).strip()
status = 'error'
return status, msg
cursor.close()
dbh.close()
return status, msg
def change_credentials(self, username, oldpwd, newpwd, db):
auth, username, error, update = self.check_credentials(username, oldpwd, db)
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
if auth:
try:
dbh = cx_Oracle.connect(username, oldpwd, dsn=dsn, newpassword=newpwd)
dbh.close()
return 'ok', "Password changed"
except Exception as e:
error = str(e).strip()
return 'error', error
if update:
try:
dbh = cx_Oracle.connect(username, oldpwd, dsn=dsn, newpassword=newpwd)
dbh.close()
return 'ok', "Password that expired was changed"
except Exception as e:
error = str(e).strip()
return 'error', error
else:
return 'error', error
def check_username(self, username, db=None):
if not db:
db = self.databases[0]
status = STATUS_OK
msg = ''
username = username.lower()
results = None
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['user'], self.conf[db]['passwd'], dsn=dsn)
cursor = dbh.cursor()
sql = """
SELECT USERNAME FROM DES_ADMIN.DES_USERS WHERE USERNAME = :username
"""
try:
results = cursor.execute(sql, username=username).fetchone()
if results:
status = STATUS_ERROR
msg = 'Username {} is unavailable. Choose a different one.'.format(username)
except Exception as e:
msg = str(e).strip()
status = STATUS_ERROR
cursor.close()
dbh.close()
return status, msg
def update_password(self, username, password):
status = STATUS_OK
msg = ''
username = username.lower()
# Sanitize inputs manually since DDL statements cannot use bind variables:
# https://cx-oracle.readthedocs.io/en/latest/user_guide/bind.html
if not self.is_alphanumeric(username):
status = STATUS_ERROR
msg = 'Invalid characters in username'
return status, msg
if not self.is_alphanumeric(password):
status = STATUS_ERROR
msg = 'Invalid characters in password'
return status, msg
for db in self.databases:
# Open an Oracle connection and get a Cursor object
try:
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['admin_user'], self.conf[db]['admin_passwd'], dsn=dsn)
# logger.info('connecting to {} with user: {}'.format(db, self.conf[db]['admin_user']))
cursor = dbh.cursor()
# Unlock account in case it is locked for some reason
status, msg = self.unlock_account(username, db)
if status != STATUS_OK:
status = STATUS_ERROR
logger.error(msg)
cursor.close()
dbh.close()
return status, msg
# If on the public interface, use a different SQL command than in the private interface
if envvars.DESACCESS_INTERFACE == 'public':
sql = """
ALTER USER {username} IDENTIFIED BY {password}
""".format(username=username, password=password)
cursor.execute(sql)
dbh.commit()
else:
logger.info('Executing RESET_PASSWORD ({},****)...'.format(username))
result = cursor.callproc('RESET_PASSWORD', [username, password])
# logger.info('Result RESET_PASSWORD: {}'.format(result))
# If the procedure calls do not throw an error, assume success
# Delete the reset token
logger.info('Deleting reset token "{}"...'.format(username))
status, msg = self.clear_reset_token(username, db)
cursor.close()
dbh.close()
except Exception as e:
status = STATUS_ERROR
msg = str(e).strip()
logger.error(msg)
cursor.close()
dbh.close()
break
return status, msg
def clear_reset_token(self, username, db=None):
if not db:
db = self.databases[0]
status = STATUS_OK
msg = ''
username = username.lower()
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['admin_user'], self.conf[db]['admin_passwd'], dsn=dsn)
cursor = dbh.cursor()
try:
# Delete the reset token
sql = """
DELETE FROM DES_ADMIN.RESET_URL WHERE USERNAME = :username
"""
cursor.execute(sql, username=username)
dbh.commit()
except Exception as e:
status = STATUS_ERROR
msg = str(e).strip()
logger.error(msg)
cursor.close()
dbh.close()
return status, msg
def unlock_account(self, username, db=None):
if not db:
db = self.databases[0]
status = STATUS_OK
msg = ''
username = username.lower()
# Sanitize inputs manually since DDL statements cannot use bind variables:
# https://cx-oracle.readthedocs.io/en/latest/user_guide/bind.html
if not self.is_alphanumeric(username):
status = STATUS_ERROR
msg = 'Invalid characters in username'
return status, msg
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['admin_user'], self.conf[db]['admin_passwd'], dsn=dsn)
cursor = dbh.cursor()
try:
if envvars.DESACCESS_INTERFACE == 'public':
sql = """
ALTER USER {} ACCOUNT UNLOCK
""".format(username)
cursor.execute(sql)
dbh.commit()
else:
logger.info('Executing UNLOCKUSER...')
result = cursor.callproc('UNLOCKUSER', [username])
logger.info('Result UNLOCKUSER: {}'.format(result))
# Delete the reset token
status, msg = self.clear_reset_token(username, db)
except Exception as e:
status = STATUS_ERROR
msg = str(e).strip()
logger.error(msg)
cursor.close()
dbh.close()
return status, msg
def validate_token(self, token, timeout=6000):
valid = False
status = STATUS_OK
msg = 'Activation token is invalid'
results = None
# Reset tokens need only be managed in one database, since we are applying updated user info and credentials to all databases
db = self.databases[0]
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['admin_user'], self.conf[db]['admin_passwd'], dsn=dsn)
cursor = dbh.cursor()
sql = """
SELECT CREATED, USERNAME FROM DES_ADMIN.RESET_URL WHERE URL = :token
"""
try:
created, username = None, None
for row in cursor.execute(sql, token=token):
created, username = row
if not created:
msg = 'Activation token is invalid'
logger.info(msg)
else:
if (dt.datetime.now() - created).seconds > timeout:
msg = 'Activation token has expired'
logger.info(msg)
else:
msg = ''
valid = True
except Exception as e:
logger.error('Error selecting reset URL')
valid = False
# msg = str(e).strip()
status = STATUS_ERROR
cursor.close()
dbh.close()
return valid, username, status, msg
def check_email(self, email, db=None):
if not db:
db = self.databases[0]
status = STATUS_OK
msg = ''
results = None
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['user'], self.conf[db]['passwd'], dsn=dsn)
cursor = dbh.cursor()
sql = """
SELECT EMAIL FROM DES_ADMIN.DES_USERS WHERE EMAIL = :email
"""
try:
results = cursor.execute(sql, email=email.lower()).fetchone()
if results:
status = STATUS_ERROR
msg = 'Email address {} is already registered.'.format(email)
except Exception as e:
msg = str(e).strip()
status = STATUS_ERROR
cursor.close()
dbh.close()
return status, msg
def create_reset_url(self, username, email=''):
url = None
firstname = None
lastname = None
status = STATUS_OK
msg = ''
username = username.lower()
# Reset tokens need only be managed in one database, since we are applying updated user info and credentials to all databases
db = self.databases[0]
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
# The admin credentials are required for the delete and insert commands
dbh = cx_Oracle.connect(self.conf[db]['admin_user'], self.conf[db]['admin_passwd'], dsn=dsn)
cursor = dbh.cursor()
try:
if email:
username_or_email = 'EMAIL'
identifier = email
else:
username_or_email = 'USERNAME'
identifier = username
# Get user profile
sql = """
SELECT USERNAME, EMAIL, FIRSTNAME, LASTNAME from DES_ADMIN.DES_USERS where {username_or_email} = :identifier
""".format(username_or_email=username_or_email)
results = cursor.execute(sql, identifier=identifier).fetchone()
if not results:
status = STATUS_ERROR
msg = 'user or email not registered.'
else:
username, email, firstname, lastname = results
# Delete any existing reset codes
status, msg = self.clear_reset_token(username, db)
now = dt.datetime.now().strftime("%Y/%m/%d %H:%M:%S")
url = uuid.uuid4().hex
sql = """
INSERT INTO DES_ADMIN.RESET_URL VALUES (:username, :url, to_date(:now , 'yyyy/mm/dd hh24:mi:ss'))
"""
cursor.execute(sql, username=username, url=url, now=now)
dbh.commit()
except Exception as e:
url = None
firstname = None
lastname = None
msg = str(e).strip()
status = STATUS_ERROR
cursor.close()
dbh.close()
return url, firstname, lastname, email, username, status, msg
def delete_user(self, username):
status = STATUS_OK
msg = ''
username = username.lower()
# Sanitize inputs manually since DDL statements cannot use bind variables:
# https://cx-oracle.readthedocs.io/en/latest/user_guide/bind.html
if not self.is_alphanumeric(username):
status = STATUS_ERROR
msg = 'Invalid characters in username'
return status, msg
# Users can only be deleted from the public DESDR database, so set db accordingly:
db = self.databases[0]
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
# The admin credentials are required for the DELETE and DROP commands
dbh = cx_Oracle.connect(self.conf[db]['admin_user'], self.conf[db]['admin_passwd'], dsn=dsn)
cursor = dbh.cursor()
try:
status, msg = self.clear_reset_token(username, db)
sql = """
DELETE FROM DES_ADMIN.DES_USERS where USERNAME = :username
"""
results = cursor.execute(sql, username=username)
sql = """
DROP USER {user} CASCADE
""".format(user=username)
results = cursor.execute(sql)
except Exception as e:
msg = str(e).strip()
status = STATUS_ERROR
cursor.close()
dbh.close()
return status, msg
def create_user(self, username, password, first, last, email, country = '', institution = '', lock=True):
status = STATUS_OK
msg = ''
username = username.lower()
# Sanitize inputs manually since DDL statements cannot use bind variables:
# https://cx-oracle.readthedocs.io/en/latest/user_guide/bind.html
if not self.is_alphanumeric(username):
status = STATUS_ERROR
msg = 'Invalid characters in username'
return status, msg
if not self.is_alphanumeric(password):
status = STATUS_ERROR
msg = 'Invalid characters in password'
return status, msg
# Users can only be created in the public DESDR database, so set db accordingly:
db = self.databases[0]
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
# The admin credentials are required for the CREATE, GRANT, and INSERT commands
dbh = cx_Oracle.connect(self.conf[db]['admin_user'], self.conf[db]['admin_passwd'], dsn=dsn)
cursor = dbh.cursor()
try:
sql = """
CREATE USER {user} IDENTIFIED BY {passwd} DEFAULT TABLESPACE USERS
""".format(user=username, passwd=password)
if lock:
sql = '{} ACCOUNT LOCK'.format(sql)
results = cursor.execute(sql)
sql = """
GRANT CREATE SESSION to {user}
""".format(user=username)
results = cursor.execute(sql)
tables = ['DES_ADMIN.CACHE_TABLES', 'DES_ADMIN.CACHE_COLUMNS']
for itable in tables:
sql = """
GRANT SELECT on {table} to {user}
""".format(table=itable, user=username)
results = cursor.execute(sql)
sql = """
INSERT INTO DES_ADMIN.DES_USERS VALUES (
:username, :first, :last, :email, :country, :institution
)
"""
results = cursor.execute(sql,
username=username,
first=first,
last=last,
email=email,
country=country,
institution=institution
)
sql = """
GRANT DES_READER to {user}
""".format(user=username)
results = cursor.execute(sql)
except Exception as e:
msg = str(e).strip()
status = STATUS_ERROR
cursor.close()
dbh.close()
return status, msg
def refresh_table_cache(self):
status = STATUS_OK
msg = ''
try:
if envvars.DESACCESS_INTERFACE == 'public':
refresh_tables_sql = """
insert into DES_ADMIN.CACHE_TABLES (TABLE_NAME)
select distinct synonym_name as table_name from all_synonyms where table_owner = 'DES_ADMIN'
"""
else:
refresh_tables_sql = """
insert into DES_ADMIN.CACHE_TABLES (TABLE_NAME)
select distinct t1.owner || '.' || t1.table_name as table_name
from all_tables t1,dba_users t2
where upper(t1.owner)=upper(t2.username)
and t1.owner not in
('XDB','SYS', 'EXFSYS' ,'MDSYS','WMSYS','ORDSYS','ORDDATA','SYSTEM',
'APEX_040200','CTXSYS','OLAPSYS','LBACSYS', 'DVSYS', 'DRIPUBLICADM', 'GSMADMIN_INTERNAL',
'DBSNMP','DRIPUBLICCATALOG','APPQOSSYS','OJVMSYS','OUTLN','AUDSYS','DBSFWUSER')
union
select distinct synonym_name as table_name from all_synonyms where table_owner = 'DES_ADMIN'
union
select distinct v1.owner || '.' || v1.view_name as table_name
from all_views v1,dba_users v2
where upper(v1.owner)=upper(v2.username)
and v1.owner not in
('XDB','SYS', 'EXFSYS' ,'MDSYS','WMSYS','ORDSYS','ORDDATA','SYSTEM',
'APEX_040200','CTXSYS','OLAPSYS','LBACSYS', 'DVSYS', 'DRIPUBLICADM', 'GSMADMIN_INTERNAL',
'DBSNMP','DRIPUBLICCATALOG','APPQOSSYS','OJVMSYS','OUTLN','AUDSYS','DBSFWUSER')
order by table_name
"""
refresh_columns_sql = """
insert into DES_ADMIN.CACHE_COLUMNS
select distinct(t.column_name) as column_name
from
all_tab_columns t , DES_ADMIN.CACHE_TABLES t2
where t.table_name=t2.table_name
"""
for db in self.databases:
# Connect to the relevant database
kwargs = {'host': self.conf[db]['host'], 'port': self.conf[db]['port'], 'service_name': db}
dsn = cx_Oracle.makedsn(**kwargs)
dbh = cx_Oracle.connect(self.conf[db]['admin_user'], self.conf[db]['admin_passwd'], dsn=dsn)
cursor = dbh.cursor()
# Empty cache tables
clean_cache_tables = 'DELETE from DES_ADMIN.CACHE_TABLES'
clean_cache_cols = 'DELETE from DES_ADMIN.CACHE_COLUMNS'
cursor.execute(clean_cache_tables)
cursor.execute(clean_cache_cols)
# Update the cache tables with table and column info
cursor.execute(refresh_tables_sql)
cursor.execute(refresh_columns_sql)
dbh.commit()
cursor.close()
dbh.close()
except Exception as e:
status = STATUS_ERROR
msg = str(e).strip()
logger.error(msg)
return status, msg