Skip to content

Commit

Permalink
Merge pull request #99 from inab/docs_module
Browse files Browse the repository at this point in the history
Docs module, with hooks usable from Sphinx
  • Loading branch information
jmfernandez committed Jun 5, 2024
2 parents 3340a88 + 1ab2ff4 commit ca522d2
Show file tree
Hide file tree
Showing 2 changed files with 126 additions and 26 deletions.
62 changes: 62 additions & 0 deletions wfexs_backend/docs_module.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-

# SPDX-License-Identifier: Apache-2.0
# Copyright 2020-2024 Barcelona Supercomputing Center (BSC), Spain
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from .wfexs_backend import WfExSBackend
from typing import TYPE_CHECKING

from .utils.licences import LicenceMatcherSingleton

if TYPE_CHECKING:
from typing import (
Sequence,
Tuple,
)

from .common import (
ContainerType,
LicenceDescription,
)

from .workflow_engines import (
WorkflowType,
)


def list_containers() -> "Sequence[ContainerType]":
wfBackend = WfExSBackend()
return wfBackend.listImplementedContainerTypes()


def list_export_plugins() -> "Sequence[str]":
wfBackend = WfExSBackend()
return wfBackend.listExportPluginNames()


def list_fetchers() -> "Sequence[Tuple[str, str, int]]":
wfBackend = WfExSBackend()
return wfBackend.describeFetchableSchemes()


def list_licences() -> "Sequence[LicenceDescription]":
licence_matcher = LicenceMatcherSingleton()
return licence_matcher.describeDocumentedLicences()


def list_workflow_engines() -> "Sequence[WorkflowType]":
wfBackend = WfExSBackend()
return wfBackend.WORKFLOW_ENGINES
90 changes: 64 additions & 26 deletions wfexs_backend/utils/passphrase_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import os
import random
import secrets
import tempfile
from typing import (
cast,
Expand Down Expand Up @@ -63,7 +64,7 @@ class RemoteWordlistResource(NamedTuple):
subend: "Optional[int]" = None


class WfExSPassphraseGenerator(FunnyPassphraseGenerator):
class WfExSPassphraseGenerator:
DEFAULT_PASSPHRASE_LENGTH: "Final[int]" = 6
WFEXS_PASSPHRASE_SCHEME: "Final[str]" = "wfexs.funny-passphrase"

Expand Down Expand Up @@ -104,6 +105,9 @@ class WfExSPassphraseGenerator(FunnyPassphraseGenerator):
],
}

MIN_RAND_CHARS: "Final[int]" = 5
MAX_RAND_CHARS: "Final[int]" = 13

def __init__(
self,
cacheHandler: "SchemeHandlerCacheHandler",
Expand All @@ -115,9 +119,24 @@ def __init__(
self.cacheHandler = cacheHandler
self.cacheDir = cacheDir

cindex_sets = self._materialize_word_sets(word_sets)
self._word_sets = word_sets
self._fungen: "Optional[FunnyPassphraseGenerator]" = None

@property
def fungen(self) -> "FunnyPassphraseGenerator":
if self._fungen is None:
cindex_sets = self._materialize_word_sets(self._word_sets)

self._fungen = FunnyPassphraseGenerator(**cindex_sets)

return self._fungen

@property
def initialized(self) -> "bool":
return self._fungen is not None

super().__init__(**cindex_sets)
def initialize(self) -> "bool":
return self.fungen != None

def _materialize_word_sets(
self, raw_word_sets: "Mapping[str, Sequence[RemoteWordlistResource]]"
Expand Down Expand Up @@ -187,29 +206,48 @@ def generate_passphrase_random(
"""
This method is needed to hook into the funny passphrase library
"""
get_random = 1 if chosen_wordlist is None else 0

wordlists_tags = self.word_set_tags()
if get_random == 0:
if not isinstance(chosen_wordlist, list):
chosen_wordlist = [cast("Union[str, int]", chosen_wordlist)]

# Validating the wordlist
for chosen in chosen_wordlist:
if chosen not in wordlists_tags:
get_random = len(chosen_wordlist)
break

if get_random > 0:
chosen_wordlist = [
wordlists_tags[random.randrange(len(wordlists_tags))]
for _ in range(get_random)
]

return self.generate_passphrase(
num=passphrase_length,
subset=cast("Sequence[Union[str, int]]", chosen_wordlist),
)
# Trying to avoid initializing
if chosen_wordlist is not None or self.initialized:
try:
get_random = 1 if chosen_wordlist is None else 0

wordlists_tags = self.fungen.word_set_tags()
if get_random == 0:
if not isinstance(chosen_wordlist, list):
chosen_wordlist = [cast("Union[str, int]", chosen_wordlist)]

# Validating the wordlist
for chosen in chosen_wordlist:
if chosen not in wordlists_tags:
get_random = len(chosen_wordlist)
break

if get_random > 0:
chosen_wordlist = [
wordlists_tags[random.randrange(len(wordlists_tags))]
for _ in range(get_random)
]

return self.fungen.generate_passphrase(
num=passphrase_length,
subset=cast("Sequence[Union[str, int]]", chosen_wordlist),
)
except Exception as e:
# If something happens, gracefully fallback
pass

# This path is followed when it is not initialized and
# no chosen wordlist is provided, potentially avoiding
# remote access, or some failure happened while fetching
random_passphrase = [
secrets.token_urlsafe(
secrets.randbelow(self.MAX_RAND_CHARS - self.MIN_RAND_CHARS + 1)
+ self.MIN_RAND_CHARS
)
for _ in range(passphrase_length)
]

return " ".join(random_passphrase)

def generate_nickname(self) -> str:
"""
Expand Down

0 comments on commit ca522d2

Please sign in to comment.