Add new script to import a CSV file with users

fix: validation for trusted urls

See merge request oliver/ivatar!207

.gitlab-ci.yml

@@ -14,6 +14,7 @@ before_script:
 
 test_and_coverage:
   stage: test
+  coverage: '/^TOTAL.*\s+(\d+\%)$/'
   script:
     - echo 'from ivatar.settings import TEMPLATES' > config_local.py
     - echo 'TEMPLATES[0]["OPTIONS"]["debug"] = True' >> config_local.py

.pre-commit-config.yaml

@@ -4,16 +4,16 @@ repos:
     hooks:
     -   id: check-useless-excludes
 -   repo: https://github.com/pre-commit/mirrors-prettier
-    rev: v2.4.0
+    rev: v2.6.2
     hooks:
     -   id: prettier
         files: \.(css|js|md|markdown|json)
 -   repo: https://github.com/python/black
-    rev: 21.9b0
+    rev: 22.3.0
     hooks:
     -   id: black
 -   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.0.1
+    rev: v4.2.0
     hooks:
     -   id: check-added-large-files
     -   id: check-ast
@@ -57,11 +57,11 @@ repos:
         types:
         - shell
 -   repo: https://github.com/asottile/blacken-docs
-    rev: v1.11.0
+    rev: v1.12.1
     hooks:
     -   id: blacken-docs
 -   repo: https://github.com/hcodes/yaspeller.git
-    rev: v7.0.0
+    rev: v8.0.1
     hooks:
     -   id: yaspeller
 

INSTALL.md

@@ -1,6 +1,6 @@
 # Installation
 
-## Prequisits
+## Prerequisites
 
 Python 3.x + virtualenv
 
@@ -10,6 +10,13 @@ Python 3.x + virtualenv
 yum install python34-virtualenv.noarch
 ```
 
+### Debian 11
+
+```
+sudo apt-get update
+sudo apt-get install git python3-virtualenv libmariadb-dev libldap2-dev libsasl2-dev
+```
+
 ## Checkout
 
 ~~~~bash
@@ -22,8 +29,8 @@ cd ivatar
 ~~~~bash
 virtualenv -p python3 .virtualenv 
 source .virtualenv/bin/activate
-pip install -r requirements.txt
 pip install pillow
+pip install -r requirements.txt
 ~~~~
 
 ## (SQL) Migrations

config.py

@@ -211,19 +211,72 @@ CACHE_RESPONSE = True
 
 # Trusted URLs for default redirection
 TRUSTED_DEFAULT_URLS = [
-    "https://ui-avatars.com/api/",
-    "http://gravatar.com/avatar/",
-    "https://gravatar.com/avatar/",
-    "http://www.gravatar.org/avatar/",
-    "https://www.gravatar.org/avatar/",
-    "https://secure.gravatar.com/avatar/",
-    "http://0.gravatar.com/avatar/",
-    "https://0.gravatar.com/avatar/",
-    "https://avatars.dicebear.com/api/",
-    "https://badges.fedoraproject.org/static/img/",
-    "http://www.planet-libre.org/themes/planetlibre/images/",
-    "https://www.azuracast.com/img/",
-    "https://reps.mozilla.org/static/base/img/remo/",
+    {
+        "schemes": [
+            "https"
+        ],
+        "host_equals": "ui-avatars.com",
+        "path_prefix": "/api/"
+    },
+    {
+        "schemes": [
+            "http",
+            "https"
+        ],
+        "host_equals": "gravatar.com",
+        "path_prefix": "/avatar/"
+    },
+    {
+        "schemes": [
+            "http",
+            "https"
+        ],
+        "host_suffix": ".gravatar.com",
+        "path_prefix": "/avatar/"
+    },
+    {
+        "schemes": [
+            "http",
+            "https"
+        ],
+        "host_equals": "www.gravatar.org",
+        "path_prefix": "/avatar/"
+    },
+    {
+        "schemes": [
+            "https"
+        ],
+        "host_equals": "avatars.dicebear.com",
+        "path_prefix": "/api/"
+    },
+    {
+        "schemes": [
+            "https"
+        ],
+        "host_equals": "badges.fedoraproject.org",
+        "path_prefix": "/static/img/"
+    },
+    {
+        "schemes": [
+            "http",
+        ],
+        "host_equals": "www.planet-libre.org",
+        "path_prefix": "/themes/planetlibre/images/"
+    },
+    {
+        "schemes": [
+            "https"
+        ],
+        "host_equals": "www.azuracast.com",
+        "path_prefix": "/img/"
+    },
+    {
+        "schemes": [
+            "https"
+        ],
+        "host_equals": "reps.mozilla.org",
+        "path_prefix": "/static/base/img/remo/"
+    }
 ]
 
 # This MUST BE THE LAST!

encrypted_proxy/README.md

@@ -1,8 +0,0 @@
-The code in here should be able to help to build up some encrypting proxy.
-
-If your app uses a lot of libravatar and therefore has to do a lot of DNS
-lookups, change your app in such a way, that it encodes the mail address,
-sends it over to the proxy, which will decrypt it, do the DNS lookup and
-return the image binary.
-
-No guarantee for this code. It's untested and just provided as example.

encrypted_proxy/lib/MyCypher.php

@@ -1,90 +0,0 @@
-<?php
-
-/**
- * Valid encryption methods AES-256-CFB 
- * Code kindly borrowed from:
- * https://github.com/arajapandi/php-python-encrypt-decrypt
- * 
- * $cypher = new MyCypher($iv);
- * $php_encrypted      = $cypher->encrypt('test');
- * $php_decrypted      = $cypher->decrypt($php_encrypted);
- */
-class MyCypher {
-
-    private $key = 'asdfa923aksadsYahoasdw998sdsads';
-    private $iv = null;
-    private $method = "AES-256-CFB";
-    private $blocksize = 32;
-    private $padwith = '`';
-
-    /*
-     * construct for cypher class - get, set key and iv
-     */
-
-    function __construct($iv, $key = null) {
-
-        if (is_string($key)) {
-            $this->key = $key;
-        }
-
-        $this->iv = $iv;
-    }
-
-    /*
-     * get hased key - if key is not set on init, then default key wil be used
-     */
-
-    private function getKEY() {
-
-        if (empty($this->key)) {
-            die('Key not set!');
-        }
-
-        return substr(hash('sha256', $this->key), 0, 32);
-    }
-
-    /*
-     * get hashed IV value - if no IV values then it throw error
-     */
-
-    private function getIV() {
-
-        if (empty($this->iv)) {
-            die('IV not set!');
-        }
-
-        return substr(hash('sha256', $this->iv), 0, 16);
-    }
-
-    /*
-     * Encrypt given string using AES encryption standard
-     */
-
-    public function encrypt($secret) {
-
-        try {
-
-            $padded_secret = $secret . str_repeat($this->padwith, ($this->blocksize - strlen($secret) % $this->blocksize));
-            $encrypted_string = openssl_encrypt($padded_secret, $this->method, $this->getKEY(), OPENSSL_RAW_DATA, $this->getIV());
-            $encrypted_secret = base64_encode($encrypted_string);
-            return $encrypted_secret;
-        } catch (Exception $e) {
-            die('Error : ' . $e->getMessage());
-        }
-    }
-
-    /*
-     * Decrypt given string using AES standard
-     */
-
-    public function decrypt($secret) {
-        try {
-            $decoded_secret = base64_decode($secret);
-            $decrypted_secret = openssl_decrypt($decoded_secret, $this->method, $this->getKEY(), OPENSSL_RAW_DATA, $this->getIV());
-            return rtrim($decrypted_secret, $this->padwith);
-        } catch (Exception $e) {
-            die('Error : ' . $e->getMessage());
-        }
-    }
-
-}

encrypted_proxy/lib/MyCypher.py

@@ -1,73 +0,0 @@
-#!/usr/bin/env python2
-#encoding: UTF-8
-
-# Code kindly borrowed from:
-# https://github.com/arajapandi/php-python-encrypt-decrypt
-
-# Python Class for AES encryption
-"""
-    Example Usage
-    enc_str = cipher.encrypt('secret')
-    enc_str = cipher.decrypt(enc_str)
-    print(enc_str); #secret
-"""
-
-from Crypto.Cipher import AES
-import base64
-import hashlib
-import sys
-
-class MyCypher:
-    # Default Key for encryption
-    rawkey      = 'asdfa923aksadsYahoasdw998sdsads'
-    method      = AES.MODE_CFB
-    blocksize   = 32  # 16, 32..etc
-    padwith     = '`'.encode('utf-8') # padding value for string  
-    
-    #lambda function for padding
-    pad         = lambda self, s: s + (self.blocksize - len(s) % self.blocksize) * self.padwith
-    
-    """
-    construct for cypher class - get, set key and iv
-    """
-    def __init__(self, iv, key=''):
-        
-        if(not key): 
-            key = self.rawkey
-            
-        self.key    = key.encode('utf-8')
-        self.iv     = iv.encode('utf-8')
-    
-    """
-    get hased key - if key is not set on init, then default key wil be used
-    """
-    def getKEY(self):
-        if(not self.key):
-            sys.exit()
-            
-        return hashlib.sha256(self.key).hexdigest()[:32]
-    
-    """
-    get hashed IV value - if no IV values then it throw error
-    """
-    def getIV(self):
-        if(not self.iv):
-            sys.exit()
-            
-        self.iv = self.iv
-        return hashlib.sha256(self.iv).hexdigest()[:16]
-    
-    """
-    Encrypt given string using AES encryption standard
-    """
-    def encrypt(self, raw):
-        cipher = AES.new(self.getKEY(), self.method, self.getIV(), segment_size=128)
-        return base64.b64encode(cipher.encrypt(self.pad(raw)))
-    
-    """
-    Decrypt given string using AES standard
-    """
-    def decrypt(self, encrypted):
-        encrypted = base64.b64decode(encrypted)
-        cipher = AES.new(self.getKEY(), self.method, self.getIV(), segment_size=128)
-        return cipher.decrypt(encrypted).rstrip(self.padwith)

encrypted_proxy/proxy.py

@@ -1,32 +0,0 @@
-#!/usr/bin/env python3
-
-import urllib.request
-import sys
-import os
-from lib.MyCypher import MyCypher
-import libravatar
-
-# Both need to be the same as in your client code that encrypts the
-# mail address
-iv = 'asdf'
-key = 'Hallo123'
-
-#sys.stderr.buffer.write(b'%s' % bytes(os.environ.get("QUERY_STRING", "No Query String in url"), 'utf-8'))
-
-cypher = MyCypher(iv = iv, key = key)
-mail = cypher.decrypt(os.environ.get('QUERY_STRING').encode('utf-8')).decode('utf-8')
-
-link = libravatar.libravatar_url(mail)
-sys.stderr.buffer.write(b'%s' % bytes(link, 'utf-8'))
-
-data = None
-with urllib.request.urlopen(link) as f:
-    data = f.read()
-
-for header in f.headers._headers:
-    if header[0] == 'Content-Type':
-        sys.stdout.buffer.write(b"%s: %s\n\n" % (bytes(header[0], 'utf-8'), bytes(header[1], 'utf-8')))
-        sys.stdout.flush()
-        break
-
-sys.stdout.buffer.write(data)

encrypted_proxy/test_code/decrypt.py

@@ -1,9 +0,0 @@
-#!/usr/bin/env python
-
-from MyCypher import MyCypher
-
-encstr = bytes('drEN/LqPBu1wJYHpN5eCjZXqVgvDEP3rZnXJt85Ma0k=', 'utf-8')
-
-cypher = MyCypher(iv = str('asdf'))
-print(cypher.decrypt(encstr))
-

encrypted_proxy/test_code/encrypt.php

@@ -1,8 +0,0 @@
-<?php
-include 'lib/MyCypher.php';
-
-$iv = 'asdf';
-$key = 'Hallo123';
-$cypher = new MyCypher($iv=$iv, $key=$key);
-$php_encrypted = $cypher->encrypt('oliver@linux-kernel.at');
-print($php_encrypted);

import_csv.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+Import a CSV - Format as follows:
+    <mailaddr>,<path_to_image>
+
+Example:
+    myuser@mydomain.tld,myphoto.jpeg
+
+This will create or update an existing user and assign the image
+to the given address.
+"""
+
+import os
+from os.path import isfile
+import sys
+from io import BytesIO
+import csv
+import django
+
+os.environ.setdefault(
+    "DJANGO_SETTINGS_MODULE", "ivatar.settings"
+)  # pylint: disable=wrong-import-position
+django.setup()  # pylint: disable=wrong-import-position
+from django.contrib.auth.models import User
+from PIL import Image
+from ivatar.settings import JPEG_QUALITY
+from ivatar.ivataraccount.models import ConfirmedEmail
+from ivatar.ivataraccount.models import Photo
+from ivatar.ivataraccount.models import file_format
+
+if len(sys.argv) < 2:
+    print("First argument to '%s' must be the path to the CSV" % sys.argv[0])
+    exit(-255)
+
+if not isfile(sys.argv[1]):
+    print("First argument to '%s' must be a path to the CSV" % sys.argv[0])
+    exit(-255)
+
+PATH = sys.argv[1]
+with open(PATH, newline="") as csvfile:
+    contactreader = csv.reader(csvfile, delimiter=",")
+    for row in contactreader:
+        mailaddr = row[0]
+        image = row[1]
+
+        if not isfile(image):
+            print("File '%s' doesn't exist - cannot add" % image)
+            continue
+
+        print("Adding: %s" % mailaddr)
+
+        (user, created) = User.objects.get_or_create(username=mailaddr)
+        if not user.confirmedemail_set.count() < 1:
+            ConfirmedEmail.objects.get_or_create(
+                email=mailaddr,
+                user=user,
+            )
+        user.save()
+        with open(image, "rb") as avatar:
+            pilobj = Image.open(avatar)
+            out = BytesIO()
+            pilobj.save(out, pilobj.format, quality=JPEG_QUALITY)
+            out.seek(0)
+            photo = None
+            if user.photo_set.count() < 1:
+                photo = Photo()
+                photo.user = user
+            else:
+                photo = user.photo_set.first()
+            photo.ip_address = "0.0.0.0"
+            photo.format = file_format(pilobj.format)
+            photo.data = out.read()
+            photo.save()
+            print("xxx: %s" % user.confirmedemail_set.first())
+            confirmed_email = user.confirmedemail_set.first()
+            confirmed_email.photo_id = photo.id
+            confirmed_email.save()

ivatar/ivataraccount/templates/profile.html

@@ -113,7 +113,7 @@
               </li>
               <li class="email-delete">
                 <button type="submit" class="nobutton" onclick="return confirm('{% trans 'Are you sure that you want to delete this email address?' %}')">
-                  Delete Email Adress
+                  Delete Email Address
                 </button>
               </li>
             </ul>
@@ -135,7 +135,7 @@
               </li>
               <li class="email-delete">
                 <button type="submit" class="nobutton" onclick="return confirm('{% trans 'Are you sure that you want to delete this email address?' %}')">
-                  Delete Email Adress
+                  Delete Email Address
                 </button>
               </li>
             </ul>

ivatar/ivataraccount/test_views.py

@@ -1830,4 +1830,63 @@ class Tester(TestCase):  # pylint: disable=too-many-public-methods
         Test if uploading export works
         """
 
-        self.client.get(reverse("upload_export"))
+        # Ensure we have data in place
+        self.test_export()
+
+        self.login()
+        self.client.get(reverse("export"))
+        response = self.client.post(
+            reverse("export"),
+            {},
+            follow=False,
+        )
+        self.assertIsInstance(response.content, bytes)
+
+        fh_gzip = gzip.open(BytesIO(response.content), "rb")
+        fh = BytesIO(response.content)
+
+        response = self.client.post(
+            reverse("upload_export"),
+            data={"not_porn": "on", "can_distribute": "on", "export_file": fh_gzip},
+            follow=True,
+        )
+        fh_gzip.close()
+        self.assertEqual(response.status_code, 200, "Upload worked")
+        self.assertContains(
+            response,
+            "Unable to parse file: Not a gzipped file",
+            1,
+            200,
+            "Upload didn't work?",
+        )
+
+        # Second test - correctly gzipped content
+        response = self.client.post(
+            reverse("upload_export"),
+            data={"not_porn": "on", "can_distribute": "on", "export_file": fh},
+            follow=True,
+        )
+        fh.close()
+
+        self.assertEqual(response.status_code, 200, "Upload worked")
+        self.assertContains(
+            response,
+            "Choose items to be imported",
+            1,
+            200,
+            "Upload didn't work?",
+        )
+        self.assertContains(
+            response,
+            "asdf@asdf.local",
+            2,
+            200,
+            "Upload didn't work?",
+        )
+
+    def test_prefs_page(self):
+        """
+        Test if preferences page works
+        """
+
+        self.client.get(reverse("user_preference"))

ivatar/test_utils.py

@@ -5,7 +5,7 @@ Test our utils from ivatar.utils
 
 from django.test import TestCase
 
-from ivatar.utils import openid_variations
+from ivatar.utils import is_trusted_url, openid_variations
 
 
 class Tester(TestCase):
@@ -45,3 +45,60 @@ class Tester(TestCase):
         self.assertEqual(openid_variations(openid3)[1], openid1)
         self.assertEqual(openid_variations(openid3)[2], openid2)
         self.assertEqual(openid_variations(openid3)[3], openid3)
+
+    def test_is_trusted_url(self):
+        test1 = is_trusted_url("https://gravatar.com/avatar/63a75a80e6b1f4adfdb04c1ca02e596c", [
+            {
+                "schemes": [
+                    "http",
+                    "https"
+                ],
+                "host_equals": "gravatar.com",
+                "path_prefix": "/avatar/"
+            }
+        ])
+        self.assertTrue(test1)
+
+        test2 = is_trusted_url("https://gravatar.com.example.org/avatar/63a75a80e6b1f4adfdb04c1ca02e596c", [
+            {
+                "schemes": [
+                    "http",
+                    "https"
+                ],
+                "host_suffix": ".gravatar.com",
+                "path_prefix": "/avatar/"
+            }
+        ])
+        self.assertFalse(test2)
+
+        # Test against open redirect with valid URL in query params
+        test3 = is_trusted_url("https://github.com/SethFalco/?boop=https://secure.gravatar.com/avatar/205e460b479e2e5b48aec07710c08d50", [
+            {
+                "schemes": [
+                    "http",
+                    "https"
+                ],
+                "host_suffix": ".gravatar.com",
+                "path_prefix": "/avatar/"
+            }
+        ])
+        self.assertFalse(test3)
+
+        test4 = is_trusted_url("https://ui-avatars.com/api/blah", [
+            {
+                "schemes": [
+                    "https"
+                ],
+                "host_equals": "ui-avatars.com",
+                "path_prefix": "/api/"
+            },
+            {
+                "schemes": [
+                    "http",
+                    "https"
+                ],
+                "host_suffix": ".gravatar.com",
+                "path_prefix": "/avatar/"
+            }
+        ])
+        self.assertTrue(test4)

ivatar/utils.py

@@ -5,6 +5,7 @@ Simple module providing reusable random_string function
 import random
 import string
 from PIL import Image, ImageDraw
+from urllib.parse import urlparse
 
 
 def random_string(length=10):
@@ -112,3 +113,42 @@ def mm_ng(
     )
 
     return image
+
+
+def is_trusted_url(url, url_filters):
+    """
+    Check if a URL is valid and considered a trusted URL.
+    If the URL is malformed, returns False.
+
+    Based on: https://developer.mozilla.org/en-US/docs/Mozilla/Add-ons/WebExtensions/API/events/UrlFilter
+    """
+    (scheme, netloc, path, params, query, fragment) = urlparse(url)
+
+    for filter in url_filters:
+        if "schemes" in filter:
+            schemes = filter["schemes"]
+
+            if scheme not in schemes:
+                continue
+
+        if "host_equals" in filter:
+            host_equals = filter["host_equals"]
+
+            if netloc != host_equals:
+                continue
+
+        if "host_suffix" in filter:
+            host_suffix = filter["host_suffix"]
+
+            if not netloc.endswith(host_suffix):
+                continue
+
+        if "path_prefix" in filter:
+            path_prefix = filter["path_prefix"]
+
+            if not path.startswith(path_prefix):
+                continue
+
+        return True
+
+    return False

ivatar/views.py

@@ -34,7 +34,7 @@ from .ivataraccount.models import ConfirmedEmail, ConfirmedOpenId
 from .ivataraccount.models import UnconfirmedEmail, UnconfirmedOpenId
 from .ivataraccount.models import Photo
 from .ivataraccount.models import pil_format, file_format
-from .utils import mm_ng
+from .utils import is_trusted_url, mm_ng
 
 URL_TIMEOUT = 5  # in seconds
 
@@ -146,7 +146,9 @@ class AvatarImageView(TemplateView):
         # Check for :// (schema)
         if default is not None and default.find("://") > 0:
             # Check if it's trusted, if not, reset to None
-            if not any(x in default for x in TRUSTED_DEFAULT_URLS):
+            trusted_url = is_trusted_url(default, TRUSTED_DEFAULT_URLS)
+
+            if not trusted_url:
                 print(
                     "Default URL is not in trusted URLs: '%s' ; Kicking it!" % default
                 )