[PATCH 7/7] Add input validation to the source spoke.

Vratislav Podzimek vpodzime at redhat.com
Tue Apr 22 10:39:54 UTC 2014


On Tue, 2014-04-15 at 14:22 -0400, David Shea wrote:
> Validate the network URL if using a network. Convert the duplicate repo
> name and invalid repo name checks into InputChecks. Add checks for
> invalid addon URLs and proxy URLs. Prevent the user from leaving the
> spoke if there is invalid data instead of simply discarding it. For the
> repo field checks, hitting Done with failed checks will change the
> selection in repoTreeView to the repo with invalid data and focus the
> problematic input box.
> ---
>  pyanaconda/ui/gui/spokes/source.glade |   3 +
>  pyanaconda/ui/gui/spokes/source.py    | 215 ++++++++++++++++++++++++++++++++--
>  2 files changed, 207 insertions(+), 11 deletions(-)
> 
> diff --git a/pyanaconda/ui/gui/spokes/source.glade b/pyanaconda/ui/gui/spokes/source.glade
> index f1db098..1783518 100644
> --- a/pyanaconda/ui/gui/spokes/source.glade
> +++ b/pyanaconda/ui/gui/spokes/source.glade
> @@ -480,6 +480,9 @@
>        <!-- column-name repo -->
>        <column type="PyObject"/>
>      </columns>
> +    <signal name="row-changed" handler="on_repoStore_row_changed" swapped="no"/>
> +    <signal name="row-deleted" handler="on_repoStore_row_deleted" swapped="no"/>
> +    <signal name="row-inserted" handler="on_repoStore_row_inserted" swapped="no"/>
>    </object>
>    <object class="AnacondaSpokeWindow" id="sourceWindow">
>      <property name="can_focus">False</property>
> diff --git a/pyanaconda/ui/gui/spokes/source.py b/pyanaconda/ui/gui/spokes/source.py
> index 0b72023..ef44d64 100644
> --- a/pyanaconda/ui/gui/spokes/source.py
> +++ b/pyanaconda/ui/gui/spokes/source.py
> @@ -25,15 +25,16 @@ import time
>  import logging
>  log = logging.getLogger("anaconda")
>  
> -import os, signal
> +import os, signal, re
>  
> -from gi.repository import GLib
> +from gi.repository import GLib, Gtk
>  
>  from pyanaconda.flags import flags
>  from pyanaconda.i18n import _, N_, CN_
>  from pyanaconda.image import opticalInstallMedia, potentialHdisoSources
>  from pyanaconda.ui.communication import hubQ
> -from pyanaconda.ui.helpers import GUIDialogInputCheckHandler, InputCheck
> +from pyanaconda.ui.helpers import GUIDialogInputCheckHandler, GUISpokeInputCheckHandler, InputCheck,\
> +        InputCheckHandler
>  from pyanaconda.ui.gui import GUIObject
>  from pyanaconda.ui.gui.spokes import NormalSpoke
>  from pyanaconda.ui.gui.categories.software import SoftwareCategory
> @@ -42,7 +43,7 @@ from pyanaconda.iutil import ProxyString, ProxyStringError, cmp_obj_attrs
>  from pyanaconda.ui.gui.utils import gtk_call_once, really_hide, really_show
>  from pyanaconda.threads import threadMgr, AnacondaThread
>  from pyanaconda.packaging import PayloadError, MetadataError
> -from pyanaconda.regexes import REPO_NAME_VALID, URL_PARSE
> +from pyanaconda.regexes import REPO_NAME_VALID, URL_PARSE, HOSTNAME_PATTERN_WITHOUT_ANCHORS
>  from pyanaconda import constants
>  
>  from blivet.util import get_mount_paths
> @@ -341,7 +342,7 @@ class IsoChooser(GUIObject):
>          if not d.startswith(constants.ISO_DIR):
>              chooser.set_current_folder(constants.ISO_DIR)
>  
> -class SourceSpoke(NormalSpoke):
> +class SourceSpoke(NormalSpoke, GUISpokeInputCheckHandler):
>      builderObjects = ["isoChooser", "isoFilter", "partitionStore", "sourceWindow", "dirImage", "repoStore"]
>      mainWidgetName = "sourceWindow"
>      uiFile = "spokes/source.glade"
> @@ -353,6 +354,7 @@ class SourceSpoke(NormalSpoke):
>  
>      def __init__(self, *args, **kwargs):
>          NormalSpoke.__init__(self, *args, **kwargs)
> +        GUISpokeInputCheckHandler.__init__(self)
>          self._currentIsoFile = None
>          self._ready = False
>          self._error = False
> @@ -611,6 +613,11 @@ class SourceSpoke(NormalSpoke):
>          self._isoChooserButton = self.builder.get_object("isoChooserButton")
>          self._origIsoChooserButton = self._isoChooserButton.get_label()
>  
> +        # Attach a validator to the URL entry. Start it as disabled, and it will be
> +        # enabled/disabled as entry sensitivity is enabled/disabled.
> +        self._urlCheck = self.add_check(self._urlEntry, self._checkURLEntry)
> +        self._urlCheck.enabled = False
> +
>          self._mirrorlistCheckbox = self.builder.get_object("mirrorlistCheckbox")
>  
>          self._noUpdatesCheckbox = self.builder.get_object("noUpdatesCheckbox")
> @@ -628,6 +635,19 @@ class SourceSpoke(NormalSpoke):
>          self._repoProxyUrlEntry = self.builder.get_object("repoProxyUrlEntry")
>          self._repoProxyUsernameEntry = self.builder.get_object("repoProxyUsernameEntry")
>          self._repoProxyPasswordEntry = self.builder.get_object("repoProxyPasswordEntry")
> +        self._repoView = self.builder.get_object("repoTreeView")
> +
> +        # Create a check for duplicate repo ids
> +        # Call InputCheckHandler directly since this check operates on rows of a TreeModel
> +        # instead of GtkEntry inputs. Updating the check is handled by the signal handlers
> +        # connected to repoStore.
> +        self._duplicateRepoCheck = InputCheckHandler.add_check(self, self._repoStore, self._checkDuplicateRepos)
> +
> +        # Create dictionaries for the checks on fields in individual repos
> +        # These checks will be added and removed as repos are added and removed from repoStore
> +        self._repoNameChecks = {}
> +        self._repoURLChecks = {}
> +        self._repoProxyChecks = {}
>  
>          # updates option container
>          self._updatesBox = self.builder.get_object("updatesBox")
> @@ -646,6 +666,7 @@ class SourceSpoke(NormalSpoke):
>          self._autodetectButton.connect("toggled", self.on_source_toggled, self._autodetectBox)
>          self._isoButton.connect("toggled", self.on_source_toggled, self._isoBox)
>          self._networkButton.connect("toggled", self.on_source_toggled, self._networkBox)
> +        self._networkButton.connect("toggled", self._updateURLEntryCheck)
>  
>          # Show or hide the updates option based on the installclass
>          if self.instclass.installUpdates:
> @@ -764,6 +785,7 @@ class SourceSpoke(NormalSpoke):
>          # something different later, we can change it.
>          self._protocolComboBox.set_active_id(PROTOCOL_MIRROR)
>          self._urlEntry.set_sensitive(False)
> +        self._updateURLEntryCheck()
>  
>          # Set up the default state of UI elements.
>          if self.data.method.method == "url":
> @@ -785,6 +807,7 @@ class SourceSpoke(NormalSpoke):
>  
>              self._urlEntry.set_sensitive(True)
>              self._urlEntry.set_text(proto[l:])
> +            self._updateURLEntryCheck()
>              self._mirrorlistCheckbox.set_active(bool(self.data.method.mirrorlist))
>              self._proxyUrl = self.data.method.proxy
>          elif self.data.method.method == "nfs":
> @@ -793,6 +816,7 @@ class SourceSpoke(NormalSpoke):
>  
>              self._urlEntry.set_text("%s:%s" % (self.data.method.server, self.data.method.dir))
>              self._urlEntry.set_sensitive(True)
> +            self._updateURLEntryCheck()
>              self.builder.get_object("nfsOptsEntry").set_text(self.data.method.opts or "")
>          elif self.data.method.method == "harddrive":
>              self._isoButton.set_active(True)
> @@ -866,6 +890,100 @@ class SourceSpoke(NormalSpoke):
>      def _sanitize_model(self, model):
>          return model.replace("_", " ")
>  
> +    # Input checks
> +
> +    # This method is shared by the checks on urlEntry and repoUrlEntry
> +    def _checkURL(self, inputcheck, combo):
> +        url_string = self.get_input(inputcheck.input_obj).strip()
> +
> +        # Don't set an error icon on empty input, but don't let the user leave the spoke
> +        if not url_string:
> +            return InputCheck.CHECK_SILENT
> +
> +        # If this is HTTP/HTTPS/FTP, use the URL_PARSE regex
> +        combo_protocol = combo.get_active_id()
> +        if combo_protocol in (PROTOCOL_HTTP, PROTOCOL_HTTPS, PROTOCOL_FTP):
> +            m = URL_PARSE.match(url_string)
> +            if not m:
> +                return _("Invalid URL")
> +
> +            # If there is a protocol in the URL, and the protocol matches the
> +            # combo box, just remove it. This makes it more convenient to paste
> +            # in URLs. It'll probably freak out people who are typing out http://
> +            # in the box themselves, but why would you do that?  Don't do that.
> +            # If the protocols don't match, complain.
> +            url_protocol = m.group('protocol')
> +            if url_protocol:
> +                if (url_protocol == 'http://' and combo_protocol == PROTOCOL_HTTP) or \
> +                        (url_protocol == 'https://' and combo_protocol == PROTOCOL_HTTPS) or \
> +                        (url_protocol == 'ftp://' and combo_protocol == PROTOCOL_FTP):
> +                    # Disable the check to block a recursive check call
> +                    inputcheck.enabled = False
> +                    inputcheck.input_obj.set_text(url_string[len(url_protocol):])
> +                    inputcheck.enabled = True
Two things here:
1) this code could live in separate method/function
2) something like 'with inputcheck_blocked:' could be useful

> +                else:
> +                    return _("Protocol in URL does not match selected protocol")
> +        elif combo_protocol == PROTOCOL_NFS:
> +            # Make sure the part before the colon looks like a hostname,
> +            # and that the path is not empty
> +            host, _colon, path = url_string.partition(':')
> +
> +            if not re.match('^' + HOSTNAME_PATTERN_WITHOUT_ANCHORS + '$', host):
> +                return _("Invalid host name")
> +
> +            if not path:
> +                return _("Remote directory is required")
> +
> +        return InputCheck.CHECK_OK
> +
> +    def _checkURLEntry(self, inputcheck):
> +        return self._checkURL(inputcheck, self._protocolComboBox)
> +
> +    def _checkRepoURL(self, inputcheck):
> +        return self._checkURL(inputcheck, self._repoProtocolComboBox)
> +
> +    # Update the check on urlEntry when the sensitity or selected protocol changes
> +    def _updateURLEntryCheck(self, *args):
> +        self._urlCheck.enabled = self._urlEntry.is_sensitive()
> +        self._urlCheck.update_check_status()
> +
> +        # Force a status update to clear any disabled errors
> +        self.set_status(self._urlCheck)
> +
> +    def _checkDuplicateRepos(self, inputcheck):
> +        repo_names = [r[REPO_OBJ].name for r in inputcheck.input_obj]
> +        if len(repo_names) != len(frozenset(repo_names)):
> +            return _("Duplicate repository names.")
> +        return InputCheck.CHECK_OK
I'd rather see a for-cycle here building a set of repo names and testing
if the currently iterated one already is in the set or not. I find that
better readable and it will go through the items only once.

> +
> +    def _checkRepoName(self, inputcheck):
> +        repo_name = self.get_input(inputcheck.input_obj).strip()
> +
> +        if not repo_name:
> +            return _("Empty repository name")
> +
> +        if not REPO_NAME_VALID.match(repo_name):
> +            return _("Invalid repository name")
> +
> +        cnames = [constants.BASE_REPO_NAME] + \
> +                 self.payload.DEFAULT_REPOS + \
> +                 [r for r in self.payload.repos if r not in self.payload.addOns]
> +        if repo_name in cnames:
> +            return _("Repository name conflicts with internal repository name.")
> +
> +        return InputCheck.CHECK_OK
> +
> +    def _checkRepoProxy(self, inputcheck):
> +        # Empty proxies are OK, as long as the username and password are empty too
> +        proxy_string = self.get_input(inputcheck.input_obj).strip()
> +        username_set = self._repoProxyUsernameEntry.is_sensitive() and self._repoProxyUsernameEntry.get_text().strip()
> +        password_set = self._repoProxyPasswordEntry.is_sensitive() and self._repoProxyPasswordEntry.get_text().strip()
> +
> +        if not (proxy_string or username_set or password_set):
> +            return InputCheck.CHECK_OK
> +
> +        return _validateProxy(proxy_string, username_set, password_set)
> +
>      # Signal handlers.
>      def on_source_toggled(self, button, relatedBox):
>          # When a radio button is clicked, this handler gets called for both
>
> @@ -876,14 +994,34 @@ class SourceSpoke(NormalSpoke):
>          self._setup_no_updates()
>  
>      def on_back_clicked(self, button):
> -        """If the user entered duplicate repo names, keep them on the screen.
> +        """If any input validation checks failed, keep the user on the screen.
>             Otherwise, do the usual thing."""
> -        ui_repo_names = [r[REPO_OBJ].name for r in self._repoStore]
>  
> -        if len(ui_repo_names) != len(frozenset(ui_repo_names)):
> +        failed_check = next(self.failed_checks, None)
> +
> +        # If the failed check is the duplicate repo check, focus the repo TreeView
> +        if failed_check == self._duplicateRepoCheck:
> +            self._repoView.grab_focus()
> +            return
> +        # If the failed check is on one of the repo fields, select the repo in the
> +        # TreeView and focus the field
> +        elif failed_check in self._repoNameChecks.values():
> +            self._repoSelection.select_path(failed_check.data.get_path())
> +            self._repoNameEntry.grab_focus()
> +            return
> +        elif failed_check in self._repoURLChecks.values():
> +            self._repoSelection.select_path(failed_check.data.get_path())
> +            self._repoUrlEntry.grab_focus()
>              return
> -        else:
> -            NormalSpoke.on_back_clicked(self, button)
> +        elif failed_check in self._repoProxyChecks.values():
> +            self._repoSelection.select_path(failed_check.data.get_path())
> +            self._repoProxyUrlEntry.grab_focus()
> +            return
> +        # Otherwise let GUISpokeInputCheckHandler figure out what to focus
> +        elif not GUISpokeInputCheckHandler.on_back_clicked(self, button):
> +            return
> +
> +        NormalSpoke.on_back_clicked(self, button)
>  
>      def on_chooser_clicked(self, button):
>          dialog = IsoChooser(self.data)
> @@ -955,6 +1093,11 @@ class SourceSpoke(NormalSpoke):
>          self._mirrorlistCheckbox.set_visible(self._http_active())
>          self._setup_no_updates()
>  
> +        # Any changes to the protocol combo box also need to update the check to see
> +        # if the protocol now matches (e.g., user puts in a ftp:// URL with http selected
> +        # in the combo box, then switches the combo box to ftp).
> +        self._updateURLEntryCheck()
> +
>      def _update_payload_repos(self):
>          """ Change the packaging repos to match the new edits
>  
> @@ -996,6 +1139,14 @@ class SourceSpoke(NormalSpoke):
>  
>              If the list has no element, clear the repo entry fields.
>          """
> +
> +        # Remove the repo checks
> +        for check in self._repoNameChecks.values() + self._repoURLChecks.values() + self._repoProxyChecks.values():
> +            self.remove_check(check)
> +        self._repoNameChecks = {}
> +        self._repoURLChecks = {}
> +        self._repoProxyChecks = {}
> +
>          self._repoStore.clear()
>          repos = self.payload.addOns
>          log.debug("Setting up repos: %s", repos)
> @@ -1127,6 +1278,16 @@ class SourceSpoke(NormalSpoke):
>          itr = self._repoSelection.get_selected()[1]
>          if not itr:
>              return
> +
> +        # Remove the input validation checks for this repo
> +        repo = self._repoStore[itr][REPO_OBJ]
> +        self.remove_check(self._repoNameChecks[repo])
> +        self.remove_check(self._repoURLChecks[repo])
> +        self.remove_check(self._repoProxyChecks[repo])
> +        del self._repoNameChecks[repo]
> +        del self._repoURLChecks[repo]
> +        del self._repoProxyChecks[repo]
> +
>          self._repoStore.remove(itr)
>          if len(self._repoStore) == 0:
>              self._clear_repo_info()
> @@ -1146,8 +1307,10 @@ class SourceSpoke(NormalSpoke):
>          repo = self._repoStore[itr][REPO_OBJ]
>          name = self._repoNameEntry.get_text().strip()
>  
> -        self._repoStore.set_value(itr, REPO_NAME_COL, name)
>          repo.name = name
> +        self._repoStore.set_value(itr, REPO_NAME_COL, name)
> +
> +        self._repoNameChecks[repo].update_check_status()
>  
>      def on_repoUrl_changed(self, *args):
>          """ proxy url or protocol changed
> @@ -1164,6 +1327,8 @@ class SourceSpoke(NormalSpoke):
>          else:
>              repo.baseurl = proto + url
>  
> +        self._repoURLChecks[repo].update_check_status()
> +
>      def on_repoMirrorlistCheckbox_toggled(self, *args):
>          """ mirror state changed
>          """
> @@ -1193,8 +1358,36 @@ class SourceSpoke(NormalSpoke):
>          username = self._repoProxyUsernameEntry.get_text().strip() or None
>          password = self._repoProxyPasswordEntry.get_text().strip() or None
>  
> +        self._repoProxyChecks[repo].update_check_status()
> +
>          try:
>              proxy = ProxyString(url=url, username=username, password=password)
>              repo.proxy = proxy.url
>          except ProxyStringError as e:
>              log.error("Failed to parse proxy - %s:%s@%s: %s", username, password, url, e)
> +
> +    def on_repoStore_row_changed(self, model, path, itr, user_data=None):
> +        self._duplicateRepoCheck.update_check_status()
> +
> +    def on_repoStore_row_deleted(self, model, path, user_data=None):
> +        self._duplicateRepoCheck.update_check_status()
> +
> +    def on_repoStore_row_inserted(self, model, path, itr, user_data=None):
> +        self._duplicateRepoCheck.update_check_status()
> +
> +        repo = model[itr][REPO_OBJ]
> +
> +        # Add checks for the repo fields
> +        # Use InputCheckHandler.add_check instead of GUISpokeInputCheckHandler.add_check since
> +        # the input fields are used by every repo, so the changed signal handler is shared by
> +        # more than one check and needs to update only the active one.
> +
> +        # It would be nice if we could store itr as the means of accessing this row later,
> +        # and GtkListStore sets GTK_TREE_MODEL_ITERS_PERSIST which is supposed to let us
> +        # do something like that, but as part of a grand practical joke the iter passed in
> +        # to this method is different from the iter used everywhere else, and is useless
> +        # once this method returns. Instead, create a TreeRowReference and work backwards
> +        # from that using paths any time we need to reference the store.
> +        self._repoNameChecks[repo] = InputCheckHandler.add_check(self, self._repoNameEntry, self._checkRepoName, Gtk.TreeRowReference.new(model, path))
> +        self._repoURLChecks[repo] = InputCheckHandler.add_check(self, self._repoUrlEntry, self._checkRepoURL, Gtk.TreeRowReference.new(model, path))
> +        self._repoProxyChecks[repo] = InputCheckHandler.add_check(self, self._repoProxyUrlEntry, self._checkRepoProxy, Gtk.TreeRowReference.new(model, path))
Please split these three lines into shorter ones.

-- 
Vratislav Podzimek

Anaconda Rider | Red Hat, Inc. | Brno - Czech Republic



More information about the anaconda-patches mailing list