0002-admin-manage-api-access-48751.patch

Nicolas Roche, 25 novembre 2020 10:53

Voir les différences: en ligne côte à côte

Subject: [PATCH 2/2] admin: manage api access (#48751)

 tests/admin_pages/test_all.py | 43 ++++++++++++++++
 wcs/admin/settings.py         | 97 ++++++++++++++++++++++++++++++++++-
 wcs/parameters.py             | 35 +++++++++++++
 wcs/qommon/publisher.py       | 16 ++++++
 4 files changed, 190 insertions(+), 1 deletion(-)

         pub.cfg['postgresql']['port'] = '5432' # from an old convert-to-sql (before #10170)
         pub.write_cfg()
         resp = app.get('/backoffice/settings/postgresql')
         assert resp.form['port'].value == '5432'
         resp = resp.form.submit()
         assert pub.cfg['postgresql']['port'] == 5432
     def test_settings_api_access(pub):
         create_superuser(pub)
         app = login(get_app(pub))
         resp = app.get('/backoffice/settings/api-access/')
         assert 'There are no API access defined at the moment.' in resp.text
         resp.forms[0]['user'] = 'jhon'
         resp.forms[0]['key'] = 'xxx'
         resp = resp.forms[0].submit('submit')
         assert resp.location == 'http://example.net/backoffice/settings/api-access/'
         resp = resp.follow()
         assert 'jhon' in resp.text
         assert pub.get_site_parameter('jhon', 'api-secrets') == 'xxx'
         resp = resp.click('jhon')
         assert resp.forms[0]['user'].value == 'jhon'
         resp.forms[0]['user'] = 'jdoe'
         resp.forms[0]['key'] = 'yyy'
         resp = resp.forms[0].submit('submit')
         assert resp.location == 'http://example.net/backoffice/settings/api-access/'
         resp = resp.follow()
         assert 'jhon' not in resp.text
         assert 'jdoe' in resp.text
         assert not pub.get_site_parameter('jhon', 'api-secrets')
         assert pub.get_site_parameter('jdoe', 'api-secrets') == 'yyy'
         resp.forms[0]['user'] = 'smith'
         resp.forms[0]['key'] = 'zzz'
         resp = resp.forms[0].submit('submit')
         resp = resp.follow()
         resp = resp.click('smith') # go to form
         resp = resp.forms[0].submit('cancel') # and cancel
         assert resp.location == 'http://example.net/backoffice/settings/api-access/'
         resp = resp.follow()
         assert 'smith' in resp.text
         resp = resp.click('smith') # go to form
         resp = resp.forms[0].submit('delete') # and delete
         assert resp.location == 'http://example.net/backoffice/settings/api-access/'
         resp = resp.follow()
         assert 'smith' not in resp.text
     def test_studio_home(pub, studio):
         create_superuser(pub)
         app = login(get_app(pub))
         resp = app.get('/backoffice/')
         assert 'studio' in resp.text
         resp = app.get('/backoffice/studio/')
         assert '../forms/' in resp.text
         assert '../cards/' in resp.text

             # restore original branding in case it has been changed
             get_publisher().cfg['branding'] = original_branding
             get_publisher().write_cfg()
             response.filter['raw'] = True
             return theme_preview
     class ApiAccessDirectory(Directory):
         _q_exports = ['', 'add']
         def get_form(self, user=None, key=None):
             form = Form(enctype='multipart/form-data')
             form.add(StringWidget, 'user', title=_('User'), required=True,
                     value=user)
             form.add(StringWidget, 'key', title=_('Key'),
                     required=True, size=70,
                      value=key)
             form.add_submit('submit', _('Save'))
             return form
         def _q_index(self):
             html_top('settings', title=_('File Types'))
             filetypes_cfg = get_cfg('filetypes', {})
             access_cfg = get_publisher().get_site_parameter('secrets', 'api')
             form = self.get_form()
             if form.get_submit() == 'submit':
                 user = form.get_widget('user').parse()
                 key = form.get_widget('key').parse()
                 get_publisher().set_site_parameter(user, key, 'api-secrets')
                 get_publisher().save_site_parameters()
                 return redirect('.')
             r = TemplateIO(html=True)
             r += htmltext('<h2>%s</h2>') % _('API access')
             if access_cfg:
                 r += htmltext('<ul class="objects-list biglist api-access">')
                 for access_id, access in enumerate(access_cfg.items()):
                     r += htmltext('<li>')
                     r += htmltext(' <a href="%s">%s <span class="extra-info"></span></a>') % (
                                     access_id,
                                     access[0],
+                                    )
                     r += htmltext('</li>')
                 r += htmltext('</ul>')
             else:
                 r += htmltext('<div class="infonotice"><p>')
                 r += _('There are no API access defined at the moment.')
                 r += htmltext('</p></div>')
             r += htmltext('<h3>%s</h3>') % _('New API access')
             r += form.render()
             return r.getvalue()
         def _q_traverse(self, path):
             get_response().breadcrumb.append(('api-access/', _('API Access')))
             return Directory._q_traverse(self, path)
         def _q_lookup(self, component):
             access_cfg = get_publisher().get_site_parameter('secrets', 'api')
             try:
                 access_id = int(component)
                 old_user, old_key = list(access_cfg.items())[access_id]
             except (ValueError, KeyError):
                 raise errors.TraversalError()
             form = self.get_form(old_user, old_key)
             form.add_submit('cancel', _('Cancel'))
             form.add_submit('delete', _('Delete'))
             if form.get_widget('cancel').parse():
                 return redirect('.')
             if form.get_submit() == 'submit':
                 del access_cfg[old_user]
                 user = form.get_widget('user').parse()
                 key = form.get_widget('key').parse()
                 get_publisher().set_site_parameter(user, key, 'api-secrets')
                 get_publisher().save_site_parameters()
                 return redirect('.')
             if form.get_submit() == 'delete':
                 del access_cfg[old_user]
                 get_publisher().save_site_parameters()
                 return redirect('.')
             get_response().breadcrumb.append((component + '/', old_user))
             html_top('settings', title=_('File Types'))
             r = TemplateIO(html=True)
             r += htmltext('<h2>%s - %s</h2>') % (_('API Access'), old_user)
             r += form.render()
             return r.getvalue()
     class SettingsDirectory(QommonSettingsDirectory):
         _q_exports = ['', 'themes', 'users',
                 'template', 'emails', 'debug_options', 'language',
                 ('import', 'p_import'), 'export', 'identification', 'sitename',
                 'sms', 'certificates', 'texts', 'install_theme',
                 'session', 'download_theme', 'smstest', 'postgresql',
                 ('admin-permissions', 'admin_permissions'), 'geolocation',
                 'theme_preview', 'filetypes',
                 ('user-template', 'user_template'),
                 ('data-sources', 'data_sources'), 'wscalls', 'logs']
                 ('data-sources', 'data_sources'), 'wscalls', 'logs',
                 ('api-access', 'api_access')]
         emails = EmailsDirectory()
         identification = IdentificationDirectory()
         users = UsersDirectory()
         texts = TextsDirectory()
         theme_preview = ThemePreviewDirectory()
         filetypes = FileTypesDirectory()
         api_access = ApiAccessDirectory()
         data_sources = NamedDataSourcesDirectory()
         wscalls = NamedWsCallsDirectory()
         logs = LoggerDirectory()
         def _q_index(self):
             html_top('settings', title = _('Settings'))
             r = TemplateIO(html=True)
-...
                         _('Session'), _('Configure session management'))
             if enabled('permissions'):
                 roles = list(Role.select())
                 if roles:
                     r += htmltext('<dt><a href="admin-permissions">%s</a></dt> <dd>%s</dd>') % (
                         _('Admin Permissions'), _('Configure access to the administration interface'))
             if enabled('api-access'):
                 r += htmltext('<dt><a href="api-access">%s</a></dt> <dd>%s</dd>') % (
                         _('API Access'), _('Configure access to the application programming interface'))
             r += htmltext('</dl></div>')
             if enabled('import-export'):
                 r += htmltext('<div class="section">')
                 r += htmltext('<h2>%s</h2>') % _('Import / Export')
                 r += htmltext('<dl>')
                 r += htmltext('<dt><a href="import">%s</a></dt> <dd>%s</dd>') % (

     # You should have received a copy of the GNU General Public License
     # along with this program; if not, see <http://www.gnu.org/licenses/>.
     from collections import OrderedDict
     import xml.etree.ElementTree as ET
     from quixote import get_publisher
     from wcs.qommon import force_str
     from wcs.qommon.xml_storage import XmlStorableObject
     class Parameters(XmlStorableObject):
         def __init__(self):
             self.parameters = OrderedDict([('options', OrderedDict())])
             self.id = None
-...
                     d[child.tag.lower()] = etree_to_dict(child)
                 if not d:
                     return element.text
                 return d
             obj.parameters = etree_to_dict(tree) or OrderedDict()
             return obj
         def export_to_xml(self, include_id=False):
             charset = get_publisher().site_charset
             root = ET.Element('parameters')
             def dict_to_etree(parent, d):
                 if not isinstance(d, dict):
                     parent.text = d
                     return
                 for key in d:
                     dict_to_etree(ET.SubElement(parent, key), d[key])
             def remove_empty_nodes(element):
                 if element.text:
                     return 1
                 nb_childs = 0
                 for child in element.getchildren():
                     if remove_empty_nodes(child) == 0:
                         element.remove(child)
                     else:
                         nb_childs += 1
                 return nb_childs
             dict_to_etree(root, self.parameters)
             remove_empty_nodes(root)
             return root
         def get(self, option, section='options'):
             d = self.parameters
             for tag in section.split('-'):
                 d = d.get(tag)
                 if not d:
                     return None
             value = d.get(option)
             return value
         def set(self, option, value, section='options'):
             d = self.parameters
             for tag in section.split('-'):
                 if not d.get(tag):
                     d[tag] = OrderedDict()
                 d = d[tag]
             d[option] = value

                 self.get_app_logger().error('failed to read site options XML file')
                 return
         def get_site_parameter(self, option, section='options'):
             if self.site_parameters is None:
                 self.load_site_parameters()
             return self.site_parameters.get(option, section)
         def set_site_parameter(self, option, value, section='options'):
             if self.site_parameters is None:
                 self.load_site_parameters()
             self.site_parameters.set(option, value, section)
         def save_site_parameters(self):
             if not self.site_parameters:
                 return
             try:
                 dump = b'<?xml version="1.0"?>\n' + self.site_parameters.export_to_xml_string()
             except:
                 self.get_app_logger().error('failed to write site options XML file')
                 return
             filename = os.path.join(self.app_dir, 'site-parameters.xml')
             storage.atomic_write(filename, dump, async_op=False)
         def get_site_option(self, option, section='options'):
             return self.get_site_parameter(option, section) or self._get_site_option(option, section)
         def set_config(self, request = None):
             self.reload_cfg()
             self.site_options = None # reset at the beginning of a request
             self.site_parameters = None
             debug_cfg = self.cfg.get('debug', {})
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0002-admin-manage-api-access-48751.patch