Implemented bulk replacing in files.

Author: KOLANICH

https_everywhere/__main__.py

@@ -0,0 +1,65 @@
+from pathlib import Path
+
+from plumbum import cli
+
+from .core import CombinedReplacerFactory
+from .core.InBufferReplacer import InBufferReplacer
+from .core.InFileReplacer import InFileReplacer
+
+from .replacers.HEReplacer import HEReplacer
+from .replacers.HSTSPreloadReplacer import HSTSPreloadReplacer
+
+
+class OurInBufferReplacer(InBufferReplacer):
+    __slots__ = ()
+    FACS = CombinedReplacerFactory(
+        {
+            "preloads": HSTSPreloadReplacer,
+            "heRulesets": HEReplacer,
+        }
+    )
+
+    def __init__(self, preloads=None, heRulesets=None):
+        super().__init__(preloads=preloads, heRulesets=heRulesets)
+
+
+class OurInFileReplacer(InFileReplacer):
+    def __init__(self, preloads=None, heRulesets=None):
+        super().__init__(OurInBufferReplacer(preloads=preloads, heRulesets=heRulesets))
+
+
+class CLI(cli.Application):
+    """HTTPSEverywhere-like URI rewriter"""
+
+
+@CLI.subcommand("bulk")
+class FileRewriteCLI(cli.Application):
+    """Rewrites URIs in files."""
+
+    def main(self, *filesOrDirs):
+        try:
+            from tqdm import tqdm
+        except ImportError:
+            def tqdm(x):
+                return x
+
+        if not filesOrDirs:
+            filesOrDirs = (".",)
+
+        for fileOrDir in tqdm(filesOrDirs):
+            fileOrDir = Path(fileOrDir).absolute()
+            if fileOrDir.is_dir():
+                files = [el for el in fileOrDir.glob("**/*") if not el.is_dir()]
+                print(files)
+            else:
+                files = [fileOrDir]
+
+            if files:
+                repl = OurInFileReplacer()
+
+                for f in files:
+                    repl(f)
+
+
+if __name__ == "__main__":
+    CLI.run()

https_everywhere/adapter.py

@@ -1,7 +1,6 @@
 from __future__ import unicode_literals
 
 import urllib3
-from urllib3.util.url import parse_url
 
 import requests
 from requests.adapters import HTTPAdapter
@@ -12,6 +11,7 @@
 from ._chrome_preload_hsts import _preload_including_subdomains
 from ._mozilla_preload_hsts import _preload_remove_negative
 from ._util import _check_in
+from .replacers.HSTSPreloadReplacer import apply_HSTS_preload
 
 PY2 = str != "".__class__
 if PY2:
@@ -154,10 +154,7 @@ def __init__(self, *args, **kwargs):
 
     def get_redirect(self, url):
         if url.startswith("http://"):
-            p = parse_url(url)
-            if _check_in(self._domains, p.host):
-                new_url = "https:" + url[5:]
-                return new_url
+            return apply_HSTS_preload(url, self._domains)
 
         return super(PreloadHSTSAdapter, self).get_redirect(url)
 

https_everywhere/core/InBufferReplacer.py

@@ -0,0 +1,28 @@
+import re
+
+from urllib3.util.url import parse_url
+
+from . import ReplaceContext, SingleURIReplacer
+
+uri_re = re.compile(
+    "(?:http|ftp):\\/\\/?((?:[\\w-]+)(?::[\\w-]+)?@)?[\\w\\.:()-]+(?:\\/[\\w\\.:()/-]*)?"
+)
+
+
+class InBufferReplacer(SingleURIReplacer):
+    __slots__ = ("singleURIReplacer",)
+    FACS = None
+
+    def __init__(self, **kwargs):
+        self.singleURIReplacer = self.__class__.FACS(**kwargs)
+
+    def _rePlaceFunc(self, m):
+        uri = m.group(0)
+        ctx = ReplaceContext(uri)
+        self.singleURIReplacer(ctx)
+        if ctx.count > 0:
+            return ctx.res
+        return uri
+
+    def __call__(self, inputStr):
+        return ReplaceContext(*uri_re.subn(self._rePlaceFunc, inputStr))

https_everywhere/core/InFileReplacer.py

@@ -0,0 +1,16 @@
+from pathlib import Path
+
+
+class InFileReplacer:
+    __slots__ = ("inBufferReplacer", "encoding")
+
+    def __init__(self, inBufferReplacer, encoding="utf-8"):
+        self.inBufferReplacer = inBufferReplacer
+        self.encoding = encoding
+
+    def __call__(self, inputFilePath):
+        t = inputFilePath.read_text(encoding=self.encoding)
+        ctx = self.inBufferReplacer(t)
+
+        if ctx.count:
+            inputFilePath.write_text(ctx.res)

https_everywhere/core/__init__.py

@@ -0,0 +1,48 @@
+from functools import partial
+
+
+class ReplaceContext:
+    __slots__ = ("res", "shouldStop", "count")
+
+    def __init__(self, res, count=0, shouldStop=False):
+        self.res = res
+        self.shouldStop = shouldStop
+        self.count = count
+
+
+class SingleURIReplacer:
+    def __init__(self, arg):
+        raise NotImplementedError
+
+    def __call__(self, ctx):
+        raise NotImplementedError
+
+
+class CombinedReplacer(SingleURIReplacer):
+    __slots__ = ("children",)
+
+    def __init__(self, children):
+        self.children = children
+
+    def __call__(self, ctx):
+        for r in self.children:
+            r(ctx)
+            if ctx.shouldStop:
+                break
+        return ctx
+
+
+class CombinedReplacerFactory:
+    __slots__ = ("args2Ctors", "ctor")
+
+    def __init__(self, args2Ctors):
+        self.args2Ctors = args2Ctors
+
+    def _gen_replacers(self, kwargs):
+        for k, v in kwargs.items():
+            c = self.args2Ctors.get(k, None)
+            if c:
+                yield c(v)
+
+    def __call__(self, **kwargs):
+        return CombinedReplacer(tuple(self._gen_replacers(kwargs)))

https_everywhere/replacers/HEReplacer.py

@@ -0,0 +1,19 @@
+from .. import _rules
+from .._rules import _get_rulesets, https_url_rewrite
+from ..core import SingleURIReplacer
+
+
+class HEReplacer(SingleURIReplacer):
+    __slots__ = ("rulesets",)
+
+    def __init__(self, rulesets):
+        if rulesets is None:
+            _get_rulesets()
+            rulesets = _rules._DATA
+        self.rulesets = rulesets
+
+    def __call__(self, ctx):
+        prevRes = ctx.res
+        ctx.res = https_url_rewrite(ctx.res, self.rulesets)
+        if prevRes != ctx.res:
+            ctx.count += 1

https_everywhere/replacers/HSTSPreloadReplacer.py

@@ -0,0 +1,29 @@
+from .._chrome_preload_hsts import _preload_including_subdomains as _get_preload_chrome
+from .._mozilla_preload_hsts import _preload_remove_negative as _get_preload_mozilla
+from ..core import SingleURIReplacer
+from urllib3.util.url import parse_url
+
+from .._util import _check_in
+
+
+def apply_HSTS_preload(url, domains):
+    p = parse_url(url)
+    if _check_in(domains, p.host):
+        new_url = "https:" + url[len(p.scheme) + 1:]
+        return new_url
+    return url
+
+
+class HSTSPreloadReplacer(SingleURIReplacer):
+    __slots__ = ("preloads",)
+
+    def __init__(self, preloads):
+        if preloads is None:
+            preloads = _get_preload_mozilla() | _get_preload_chrome()
+        self.preloads = preloads
+
+    def __call__(self, ctx):
+        prevRes = ctx.res
+        ctx.res = apply_HSTS_preload(ctx.res, self.preloads)
+        if prevRes != ctx.res:
+            ctx.count += 1

https_everywhere/replacers/__init__.py

@@ -63,4 +63,7 @@
     classifiers=classifiers.splitlines(),
     tests_require=["unittest-expander", "lxml", "tldextract", "regex"],
     # lxml is optional, needed for testing upstream rules
+    entry_points = {
+        "console_scripts": ["pyhttpeverywhere = https_everywhere.__main__:CLI"]
+    }
 )