Added files.

flask/floritiweb/lib/forms/.gitignore

@@ -0,0 +1 @@
+__pycache__/

flask/floritiweb/lib/forms/handle_params/__init__.py

@@ -0,0 +1,215 @@
+import re
+from .field import Field, ValidationError
+import sys
+
+class HandleParams:
+
+    def __init__(
+            self, params={}, sequence=None, ignored_fields=None, defaults=None, altered_fields=None
+        ):
+        self.sequence = list(sequence) if sequence else [] 
+        self.unrecognized = []
+        self.required_fields = set()
+        self.altered_fields = dict(altered_fields) if altered_fields else {}
+        self.defaults = dict(defaults) if defaults else {}
+        for key in self.defaults.keys():
+            if len(params.get(key,'') or '') > 0:
+                self.defaults[key] = params[key]
+        self.fields = {}
+        self.ignored_fields = set(ignored_fields or [])
+
+    def add_field(self, name, properties, **alter_cfg):
+        
+        properties['name'] = name
+
+        variant = self.altered_fields.get(name)
+        if variant is not None:
+            if variant in alter_cfg:
+                cfg = alter_cfg[variant]
+                if not isinstance(cfg, dict):
+                    raise RuntimeError(f'{name}: Not a dictionary for variant {variant}: {cfg}')
+                alter_cfg = dict(cfg)
+            else:
+                # A variant is requested that is not known / supported.
+                known_keys = ','.join(alter_cfg)
+                raise RuntimeError(f"Field-def {name}: No variant {variant} (keys: {known_keys})")
+        else:
+            alter_cfg = {}
+ 
+        for attr, acfg in alter_cfg.items() :
+            if attr.startswith('wrap_'):
+                attr = attr[5:]
+                wrap = True
+            else:
+                wrap = False
+            orig = properties.get(attr)
+            properties[attr] = lambda orig, *args, **kwargs: acfg(orig, *args, **kwargs) if wrap else acfg
+        if properties.pop('required', False) == True:
+            self.required_fields.add(name)
+
+        field = Field(**properties) 
+
+        default = self.defaults.get(name)
+        if default is not None:
+            default = filter_value(field.apply_filters, name, default)
+            try:
+                if field.validate(default): field.value = default 
+            except RuntimeError as e: # HandleParams.ValidationError raised spurious errors when
+                if len(default) > 0:  # used flask is used with uwsgi
+                    raise
+
+        if hasattr(field, 'digest_key') or field.multiple:
+            self.sequence.append(name)
+
+        self.fields[name] = field
+
+    def validate_params(self, params, set_invalid_value=True):
+        success = True
+
+        ignore = self.ignored_fields
+        validate_seq = [*self.sequence]
+        defer_validations = set(validate_seq)
+
+        tmp = {}
+
+        ## raise RuntimeError(type(params))
+
+        for fname in self.required_fields:
+            if len(params.get(fname, '')) == 0:
+                params[fname] = None
+
+        fieldlist = open('/tmp/validated_fields.txt','w')
+        def paramsiter():
+            for fldname, fldvalue in params.items():
+                if isinstance(fldvalue, list):
+                    for v in fldvalue:
+                        yield fldname, v
+                else:
+                    yield fldname, fldvalue
+        
+        ## raise RuntimeError(list(paramsiter()))        
+        ## raise RuntimeError(list(paramsiter()))
+        ## raise RuntimeError(params.items())
+        ## raise RuntimeError("defer_validations", list(defer_validations))
+        ## if "foerderung" in defer_validations:
+        ##     raise RuntimeError("foerderung in defer_valitations.")
+        ## else:
+        ##     raise RuntimeError("foerderung not in defer_validations.")
+        ## for fldname, fldvalue in params.items():
+        ##     raise RuntimeError(params.items())
+        ##     if fldname == 'foerderung':
+        ##         raise RuntimeError("type", type(fldvalue), fldvalue)
+        ##     if fldname == 'foerderung' and len(fldvalue) > 1:
+        ##         raise RuntimeError(fldvalue)
+        for fldname, fldvalue in paramsiter():
+            ## raise RuntimeError(fldname)
+            ## if len(fldvalue) > 1:
+            ##     raise RuntimeError(fldvalue)
+            ## if fldname == 'foerderung':
+            ##     raise RuntimeError(fldname + ' ' + fldvalue, type(fldvalue))
+            if fldvalue is not None and len(fldvalue) == 0:
+                continue
+            if ( fldname not in defer_validations
+                 and fldname in self.fields
+               ):
+                ## raise RuntimeError(fldname)
+                ## if fldname == 'foerderung':
+                ##     raise RuntimeError('test')
+                ##     raise RuntimeError(fldname + ' ' + fldvalue)
+                field = self.fields[fldname]
+                value = filter_value(field.apply_filters, fldname, fldvalue)
+                try:
+                    this_success = field.validate(value)
+                    ## von debugging sessions
+                    if fldname == 'dsgvochecked' and this_success is False:
+                        ## raise RuntimeError("this_success: " + this_success)
+                        raise RuntimeError(value)
+
+                except HandleParams.ValidationError as e:
+                    field.error_msg = str(e)
+                    field.is_valid = False
+                    print(fldname, file=fieldlist)
+                    this_success = False
+                else:
+                    field.is_valid = True
+                take_this_value = value is not None if set_invalid_value else this_success
+                if take_this_value:
+                    field.value = value
+                if not(success and this_success):
+                    success = False
+
+            elif not fldname in ignore:
+                #if hasattr(self.fields[fldname], 'digest_key'):
+                #    raise RuntimeError(fldname)
+                #elif fldname.startswith("nachweise"):
+                #    raise RuntimeError(fldname)
+                ## raise RuntimeError("validate_seq: ", validate_seq)
+                for fname in validate_seq: 
+                    key_tmpl = {}
+                    field = self.fields[fname]
+
+                    if hasattr(field, 'digest_key'):
+                        key_tmpl = field.digest_key(fldname)
+                        if not key_tmpl: continue
+                        value = filter_value(field.apply_filters, fname, fldvalue)
+                        key_tmpl['value'] = value
+                        value = key_tmpl
+                        tmp.setdefault(fname, {})
+                        slot = tmp[fname]
+
+                    elif fldname == fname:
+                        value = filter_value(field.apply_filters, fname, fldvalue)
+                        slot = tmp
+                        ## raise RuntimeError("slot: ",slot)
+                    else:
+                        continue
+
+                    if field.multiple:
+                        slot.setdefault(fldname, [])
+                        slot[fname].append(value)
+                        ## raise RuntimeError(fldname + ' ' + value)
+                    else:
+                        slot[fldname] = value
+
+        ## raise RuntimeError(tmp) 
+        fullfilled = False
+        for fldname, fldvalue in tmp.items():
+            field = self.fields[fldname]
+            try:
+                ##raise RuntimeError("fldvalue2: ", fldvalue)
+                this_success = field.validate(fldvalue)
+            except HandleParams.ValidationError as e:
+                field.error_msg = str(e)
+                field.is_valid = False
+                this_success = False
+            else:
+                field.is_valid = True
+            take_this_value = fldvalue is not None if set_invalid_value else this_success
+            if take_this_value:
+                field.value = fldvalue
+                fullfilled = True
+            if not(success and this_success):
+                success = False
+        #if not fullfilled:
+        #    self.unrecognized.append(fldname)
+        #    continue
+    ##raise RuntimeError(tmp)
+        return success
+
+ADD_FILTERS = {}
+
+def filter_value(filters_to_apply, name, value):
+    if isinstance(value, str):
+        value = re.sub(r'[\;\"\(\)\{\}\\\`\|\%\r\<\>\&\0]', '', value)
+        for fname in filters_to_apply:
+            if not fname in ADD_FILTERS:
+                raise RuntimeError(f"No filter {fname}")
+            if not isinstance(value, list):
+                value = [list]
+            for v in value:
+                try:
+                    value = FILTERS[fname](v)
+                except HandleParams.ValidationError as e:
+                    field.error_msg = str(e)
+                    return
+        return value

flask/floritiweb/lib/forms/handle_params/field.py

@@ -0,0 +1,45 @@
+
+CALLBACKS = {
+    'validate': lambda self, value: True,
+    'render': lambda self, *args, **kwargs: '',
+    'digest_key': None,
+    'rendervalue': lambda self, *args, **kwargs: self.value,
+    'initializer': None,
+}
+
+class Field:
+    
+    def wrap_callback(self, cb):
+        if cb is None:
+            return
+        def func(*args, **kwargs):
+            return cb(self, *args, **kwargs)
+        return func
+       
+    def __init__(self, *, is_multiple=False, **callbacks):
+        self.apply_filters = []
+        for cb in CALLBACKS.keys():
+            func = self.wrap_callback(
+                callbacks.get(cb, CALLBACKS[cb])
+            )
+            if func is not None:
+                setattr(self, cb, func)
+        self.initializer = callbacks.get('initializer', lambda: '')
+        #self.digest_key = callbacks.get('digest_key', None)
+        self.error_msg = ''
+        self.multiple = is_multiple
+        self.options_map = {}
+        self.is_valid = None
+
+    @property
+    def value(self):
+        if not hasattr(self, '_value'):
+            self._value = self.initializer()
+        return self._value
+
+    @value.setter
+    def value(self, value):
+        self._value = value
+
+class ValidationError(RuntimeError):
+    pass

flask/floritiweb/lib/forms/handle_params/timewindowcheck.py

@@ -0,0 +1,93 @@
+from datetime import datetime
+from . import ValidationError
+import hmac, os, sqlite3
+
+DB_FILENAME = "/var/tmp/www/forms-sent-timestamps.db"
+SECRET = b"Dieser Zeitstempel wurde von uns gepraegt. -- UB/fh 24.5.24"
+TIME_FORMAT = "%Y-%m-%d %H:%M:%S.%f"
+MIN_DURATION = 10        # Sekunden
+
+REFUSE_NTH_TIMESTAMP_REUSE = 3 # Diese Werte werden nur bei der Erstellung einer
+MAX_REQUESTS_PER_IP = 10       # neuen SQLite3-Datenbank berücksichtigt.
+
+def digest_of(value):
+    return hmac.new(SECRET, value.encode("utf-8"), "sha256").hexdigest()
+
+def signed_fresh_timestamp_check(remote_ip, error=ValidationError):
+
+    def render_timestamp(self, name):
+        value = datetime.now().strftime(TIME_FORMAT)
+        value = "{}/{}".format(value, digest_of(value))
+        return f'<input name="{name}" type="hidden" value="{value}">'
+
+    def rendervalue(self, name):
+        value = datetime.now().strftime(TIME_FORMAT)
+        return "{}/{}".format(value, digest_of(value))
+
+    def timestamp_checker(field, value, remote_ip=remote_ip):
+        if value is None: raise error("not passed")
+        value, digest = value.split("/")
+        if not hmac.compare_digest(digest, digest_of(value)):
+            raise error("invalid")
+        timediff = datetime.now() - datetime.strptime(value, TIME_FORMAT)
+        db = init_db()
+        try:
+            c = db.cursor()
+            c.execute("""
+                INSERT OR IGNORE into timestamps (timestamp, digest, remote_ip) values (?, ?, ?)
+            """, (value, digest, remote_ip));
+            c.execute("UPDATE timestamps SET used=used+1 WHERE timestamp=? AND remote_ip=?",
+                 (value, remote_ip)
+            )
+            db.commit()
+        except sqlite3.IntegrityError:
+            raise error("overused")
+        else:
+            if c.rowcount == 0:
+                raise error("missing")
+            elif timediff.days > 0:
+                raise error("outdated")
+            elif timediff.seconds < MIN_DURATION:
+                raise error(f"fast_as_spambot")
+        finally:
+            db.close()
+        
+        return True
+
+    return {
+        "validate": timestamp_checker,
+        "render": render_timestamp,
+        "rendervalue": rendervalue,
+        "required": True
+    }
+
+
+def init_db(filename=DB_FILENAME):
+    os.makedirs(os.path.dirname(filename), exist_ok=True)
+    conn = sqlite3.connect(filename)
+    if (not os.path.isfile(filename) or os.stat(filename).st_size == 0):
+        cursor = conn.cursor()
+        cursor.executescript(f"""
+            CREATE TABLE timestamps (
+                timestamp PRIMARY KEY,
+                digest,
+                remote_ip,
+                reqtime CURRENT_TIMESTAMP,
+                used DEFAULT 0 CHECK(used < {REFUSE_NTH_TIMESTAMP_REUSE})
+            );
+            CREATE TRIGGER block_hyperactive_ip
+                BEFORE INSERT ON timestamps
+                WHEN EXISTS (
+                    SELECT 1
+                    FROM timestamps
+                    WHERE remote_ip=NEW.remote_ip
+                    GROUP by remote_ip
+                    HAVING count(timestamp)={MAX_REQUESTS_PER_IP}
+                )
+                BEGIN
+                    SELECT RAISE(FAIL, "no more than {MAX_REQUESTS_PER_IP} requests per IP");
+                    -- Still, cron'd housekeeping is required
+                END;
+            """)
+        conn.commit()
+    return conn

flask/floritiweb/lib/forms/mailform.py

@@ -0,0 +1,237 @@
+from .handle_params import HandleParams, timewindowcheck
+from email.message import EmailMessage
+from subprocess import Popen, PIPE
+import sys
+import datetime
+from textwrap import TextWrapper
+from io import StringIO
+import os
+aa = open('/web/etc/allowed.adress')
+
+addresses = {}
+for line in aa:
+    short, longname, mailAddress = line.split(";")
+    addresses[short] = (longname, mailAddress.strip())
+
+try:
+    aa.close()
+except AttributeError:
+    pass
+
+class ValidationError(RuntimeError):
+    def __init__(self, message):
+        self.langs = { 'de': message, 'en': message }
+
+    def __str__(self):
+        return self.langs['de']
+
+HandleParams.ValidationError = ValidationError
+
+class MultiLangExc(ValidationError):
+    def __init__(self, **langs):
+        self.langs = langs
+
+timewindowcheck.init_db()
+
+def init(params, altfields, defaults, check_remote_ip='0.0.0.0'):
+
+    hp = HandleParams(altered_fields=altfields, params=params, defaults=defaults)
+
+    def render_text(field, **args):
+        value = field.value
+        attr = attributes_to_string(args)
+        return f'<input style="width:75%;" type="text" value="{value}" {attr}>'
+
+    def render_textfield(field,**args):
+        value = field.value
+        attr = attributes_to_string(args)
+        return f'<textarea style="width: 75%; min-height: 12em;" {attr}>{value}</textarea>'
+
+    def render_checkbox(field, **args):
+        attr = attributes_to_string(args)
+        return f'<input type="checkbox" {attr} >'
+
+    mandatory_text = { 'de': "obligatorisch", 'en': "mandatory" }
+
+    def render_local_time(field,**args):
+        return str(datetime.datetime.now())
+
+    def must_be_nonempty(field, value):
+        if value is None or len(value) == 0:
+            raise MultiLangExc(**mandatory_text)
+        return True
+
+    def must_be_nonempty_then_validate(inner_val, field, value):
+        if len(value) == 0:
+            raise MultiLangExc(**mandatory_text)
+        return inner_val(self, value)
+
+    def validate_honeypot_field(field,value):
+        if len(value):
+            raise ValidationError(
+                "Sie haben ein verstecktes Feld ausgefüllt"
+                "(Honigtopf für Spambots)"
+            )
+        return True
+
+    def pass_longname(field, name):
+        value = field.value
+        try:
+            longname = addresses[value][0]
+        except KeyError:
+            longname = "Non-existent Enterprises Ltd (will forward)"
+            value = "ub"
+        return longname + f"<input type='hidden' name='adr' value='{value}'>"
+
+    def check_is_within_last24h():
+        pass
+
+    bei_allen_feldern_mgl = {
+        "mandatory": {
+            "required": True,
+            "wrap_validate": must_be_nonempty_then_validate,
+        }
+    }
+
+    hp.add_field("realname", {
+        "render":render_text
+        }, 
+        **bei_allen_feldern_mgl
+    )
+
+    hp.add_field("subject", {
+        "render":render_text, 
+        "required": True, 
+    })
+
+    hp.add_field("name", {
+        "validate": validate_honeypot_field, 
+        "render":render_text
+        }, 
+    )
+
+    hp.add_field("body", {
+        "render":render_textfield, 
+        "required": True, 
+    })
+
+    hp.add_field("mailadr", {
+        "render":render_text
+    })
+
+    hp.add_field("wohnort", {
+        "render":render_text
+    })
+
+    hp.add_field("adr", {
+        "render":pass_longname
+    })
+   
+    def twc_error(which):
+        msgs = {
+            "not passed": {
+                "de": "Spamprüfung gescheitert: kein timewindowcheck-Wert übergeben",
+                "en": "Spam detection failed: No field 'timewindowcheck' passed"
+            },
+            "invalid": { "de": "Zeitstempel und Digest ungültig", "en": "Timestamp and digest invalid" },
+            "missing": { "de": "Zeitstempel oder IP-Adresse nicht ermittelbar", "en": "Timestamp or IP address not indicated" },
+            "overused": { "de": "Der Zeitstempel wurde zu häufig verwendet", "en": "The timestamp has been used too often" },
+            "outdated": { "de": "Die Formularabfrage ist zu alt. Bitte laden Sie die Seite neu.", "en": "Form request is too old. Please reload the page." },
+            "fast_as_spambot": { "de": "Sie sind so schnell wie ein Spambot (oder schneller). Bitte lassen Sie sich vor dem Senden etwas mehr Zeit.", "en": "You are as fast as a spambot (or faster). Please have a breath or two before sending." },
+        }
+        return MultiLangExc(**msgs[which])
+
+    hp.add_field(
+        'timewindowcheck',
+        timewindowcheck.signed_fresh_timestamp_check(check_remote_ip, twc_error)
+    )
+
+    hp.add_field("dsgvochecked", {
+        "render":render_checkbox, 
+        "rendervalue":render_local_time, 
+        "required": True, 
+        "validate": must_be_nonempty
+    })
+    
+    return hp
+
+def process(hp):
+
+    form = {
+        "adr": "adr",
+        "subject": "subject",
+        "realname": "realname",
+        "body": "body",
+        "ort": "wohnort",
+        "mymail": "mailadr",
+        "dsgvo_ts": "dsgvochecked",
+    }
+
+    for k,v in form.items():
+        form[k] = hp.fields[v].rendervalue() or hp.fields[v].value
+
+    adresse_formular = addresses[ hp.fields["adr"].value ][1]
+
+    # Mail bzw. Request rausschicken
+    send_mail(form, adresse_formular)
+
+    return form
+
+
+def attributes_to_string(args):
+    attrs = []
+    for key, value in args.items():
+        attrs.append(f'{key}="{value}"')
+    return " ".join(attrs)
+
+
+def send_mail(data, adr):
+
+    _w = TextWrapper(width=67, subsequent_indent=" "*13)
+    def wrap(label, slot):
+        _w.initial_indent = label+" "*(11-len(label))+': '
+        return _w.wrap(data[slot])
+
+    form = {
+        "Adressat": "adr",
+        "Betreff": "subject",
+        "Inhalt": "body",
+        "Absender": "realname",
+        "Wohnort": "ort",
+        "Antwort an": "mymail",
+        "DSE z.k.g.": "dsgvo_ts"
+    }
+
+    with open('/tmp/data_test.txt', 'w') as outfile:
+        outfile.write(str(data)) 
+
+    userinput = list() 
+    for label,slot in form.items():
+        userinput.append('\n'.join(wrap(label,slot)))
+
+    text = ('' if data['mymail'] else """
+-----------------------------------------------------------------
+Bitte nicht die REPLY-Funktion verwenden. Es wurde keine Absende-
+adresse angegeben.
+Dieser Hinweis kann geloescht werden.
+-----------------------------------------------------------------
+""") + "\n".join(userinput)
+
+    mail = EmailMessage()
+    mail['From'] = 'WWW-Formular <noreply@example.invalid'
+    #mail['From'] = data['mymail'] or 'WWW-Formular'
+    mail['To'] = adr
+    del mail['Subject']
+    mail['Subject'] = data['subject']
+    mail.add_header('Content-Type', 'text/plain;charset=utf-8')
+    mail.set_payload(text)
+
+    mail_string = str(mail)
+   
+    sml = Popen(['/usr/sbin/sendmail', '-t', '-oi'], stdin=PIPE)
+    #err = sml.communicate(mail.as_bytes(policy=mail.policy.clone(linesep='\r\n')))
+    with open('/var/tmp/mail_string_encode.txt', 'w') as file:
+        print(mail_string, file=file)
+
+    err = sml.communicate(mail_string.encode('utf8'))
+ 

flask/floritiweb/lib/forms/test_handle_params.py

@@ -0,0 +1,37 @@
+from handle_params import HandleParams, ValidationError
+import pytest, re
+
+def regex_validator(regex, error_message):
+
+    def _validate(_, value):
+        if len(value) > 0:
+            m = re.search(regex, value)
+        else:
+            return
+
+        if m:
+            return value
+        else:
+            raise ValidationError(error_message)
+
+    return _validate
+
+
+def test_hp_init():
+    hp = HandleParams()
+
+
+def test_hp_with_text_field():
+    hp = HandleParams()
+    hp.add_field('name', {
+        'validate': regex_validator(
+            r'^([A-Z]\S+\.?) ([A-Z]\S+\.? )?([A-Z]\S+\.?)$',
+            'Does not look like a name'
+        )
+    })
+    hp.validate_params({
+        'name': 'Florian Heß'
+    })
+    assert hp.fields['name'].value == 'Florian Heß'
+    with pytest.raises(ValidationError):
+        assert hp.validate_params({ 'name': '123' })