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
|
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import datetime
import logging
import os
import re
import tempfile
from lxml import html
import odoo
import odoo.modules.registry
from odoo import http
from odoo.http import content_disposition, dispatch_rpc, request, Response
from odoo.service import db
from odoo.tools.misc import file_open, str2bool
from odoo.tools.translate import _
from odoo.addons.base.models.ir_qweb import render as qweb_render
_logger = logging.getLogger(__name__)
DBNAME_PATTERN = '^[a-zA-Z0-9][a-zA-Z0-9_.-]+$'
class Database(http.Controller):
def _render_template(self, **d):
d.setdefault('manage', True)
d['insecure'] = odoo.tools.config.verify_admin_password('admin')
d['list_db'] = odoo.tools.config['list_db']
d['langs'] = odoo.service.db.exp_list_lang()
d['countries'] = odoo.service.db.exp_list_countries()
d['pattern'] = DBNAME_PATTERN
# databases list
try:
d['databases'] = http.db_list()
d['incompatible_databases'] = odoo.service.db.list_db_incompatible(d['databases'])
except odoo.exceptions.AccessDenied:
d['databases'] = [request.db] if request.db else []
templates = {}
with file_open("web/static/src/public/database_manager.qweb.html", "r") as fd:
templates['database_manager'] = fd.read()
with file_open("web/static/src/public/database_manager.master_input.qweb.html", "r") as fd:
templates['master_input'] = fd.read()
with file_open("web/static/src/public/database_manager.create_form.qweb.html", "r") as fd:
templates['create_form'] = fd.read()
def load(template_name):
fromstring = html.document_fromstring if template_name == 'database_manager' else html.fragment_fromstring
return (fromstring(templates[template_name]), template_name)
return qweb_render('database_manager', d, load)
@http.route('/web/database/selector', type='http', auth="none")
def selector(self, **kw):
if request.db:
request.env.cr.close()
return self._render_template(manage=False)
@http.route('/web/database/manager', type='http', auth="none")
def manager(self, **kw):
if request.db:
request.env.cr.close()
return self._render_template()
@http.route('/web/database/create', type='http', auth="none", methods=['POST'], csrf=False)
def create(self, master_pwd, name, lang, password, **post):
insecure = odoo.tools.config.verify_admin_password('admin')
if insecure and master_pwd:
dispatch_rpc('db', 'change_admin_password', ["admin", master_pwd])
try:
if not re.match(DBNAME_PATTERN, name):
raise Exception(_('Houston, we have a database naming issue! Make sure you only use letters, numbers, underscores, hyphens, or dots in the database name, and you\'ll be golden.'))
# country code could be = "False" which is actually True in python
country_code = post.get('country_code') or False
dispatch_rpc('db', 'create_database', [master_pwd, name, bool(post.get('demo')), lang, password, post['login'], country_code, post['phone']])
credential = {'login': post['login'], 'password': password, 'type': 'password'}
request.session.authenticate(name, credential)
request.session.db = name
return request.redirect('/odoo')
except Exception as e:
_logger.exception("Database creation error.")
error = "Database creation error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
@http.route('/web/database/duplicate', type='http', auth="none", methods=['POST'], csrf=False)
def duplicate(self, master_pwd, name, new_name, neutralize_database=False):
insecure = odoo.tools.config.verify_admin_password('admin')
if insecure and master_pwd:
dispatch_rpc('db', 'change_admin_password', ["admin", master_pwd])
try:
if not re.match(DBNAME_PATTERN, new_name):
raise Exception(_('Houston, we have a database naming issue! Make sure you only use letters, numbers, underscores, hyphens, or dots in the database name, and you\'ll be golden.'))
dispatch_rpc('db', 'duplicate_database', [master_pwd, name, new_name, neutralize_database])
if request.db == name:
request.env.cr.close() # duplicating a database leads to an unusable cursor
return request.redirect('/web/database/manager')
except Exception as e:
_logger.exception("Database duplication error.")
error = "Database duplication error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
@http.route('/web/database/drop', type='http', auth="none", methods=['POST'], csrf=False)
def drop(self, master_pwd, name):
insecure = odoo.tools.config.verify_admin_password('admin')
if insecure and master_pwd:
dispatch_rpc('db', 'change_admin_password', ["admin", master_pwd])
try:
dispatch_rpc('db', 'drop', [master_pwd, name])
if request.session.db == name:
request.session.logout()
return request.redirect('/web/database/manager')
except Exception as e:
_logger.exception("Database deletion error.")
error = "Database deletion error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
@http.route('/web/database/backup', type='http', auth="none", methods=['POST'], csrf=False)
def backup(self, master_pwd, name, backup_format='zip'):
insecure = odoo.tools.config.verify_admin_password('admin')
if insecure and master_pwd:
dispatch_rpc('db', 'change_admin_password', ["admin", master_pwd])
try:
odoo.service.db.check_super(master_pwd)
if name not in http.db_list():
raise Exception("Database %r is not known" % name)
ts = datetime.datetime.utcnow().strftime("%Y-%m-%d_%H-%M-%S")
filename = "%s_%s.%s" % (name, ts, backup_format)
headers = [
('Content-Type', 'application/octet-stream; charset=binary'),
('Content-Disposition', content_disposition(filename)),
]
dump_stream = odoo.service.db.dump_db(name, None, backup_format)
response = Response(dump_stream, headers=headers, direct_passthrough=True)
return response
except Exception as e:
_logger.exception('Database.backup')
error = "Database backup error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
@http.route('/web/database/restore', type='http', auth="none", methods=['POST'], csrf=False, max_content_length=None)
def restore(self, master_pwd, backup_file, name, copy=False, neutralize_database=False):
insecure = odoo.tools.config.verify_admin_password('admin')
if insecure and master_pwd:
dispatch_rpc('db', 'change_admin_password', ["admin", master_pwd])
try:
data_file = None
db.check_super(master_pwd)
with tempfile.NamedTemporaryFile(delete=False) as data_file:
backup_file.save(data_file)
db.restore_db(name, data_file.name, str2bool(copy), neutralize_database)
return request.redirect('/web/database/manager')
except Exception as e:
error = "Database restore error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
finally:
if data_file:
os.unlink(data_file.name)
@http.route('/web/database/change_password', type='http', auth="none", methods=['POST'], csrf=False)
def change_password(self, master_pwd, master_pwd_new):
try:
dispatch_rpc('db', 'change_admin_password', [master_pwd, master_pwd_new])
return request.redirect('/web/database/manager')
except Exception as e:
error = "Master password update error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
@http.route('/web/database/list', type='json', auth='none')
def list(self):
"""
Used by Mobile application for listing database
:return: List of databases
:rtype: list
"""
return http.db_list()
|
