WRN-1934: Add method in Subversion to list all plugins with a new ver…

…sion added since a specific date.

openwebvulndb/common/vcs.py

@@ -21,11 +21,12 @@
 from os.path import join
 from os import mkdir, walk, rmdir, remove
 from contextlib import contextmanager
+from urllib.parse import urljoin, urlparse, urlunparse
+import re
+from datetime import datetime
 
 from .errors import ExecutionFailure, DirectoryExpected
 from .logs import logger
-from urllib.parse import urljoin, urlparse, urlunparse
-import re
 
 
 class Workspace:
@@ -84,7 +85,7 @@ async def ls(self, url):
         except asyncio.TimeoutError:
             raise ExecutionFailure('Timeout reached')
 
-    async def read_lines(self, command):
+    async def read_lines(self, command, *, ignore_errors=False):
         process = await create_subprocess_exec(
             *command,
             loop=self.loop,
@@ -101,7 +102,7 @@ async def read_lines(self, command):
 
         # No need to wait for a long time, we're at EOF
         code = await process.wait()
-        if code == 0:
+        if code == 0 or ignore_errors:
             return out
 
         raise ExecutionFailure("Listing failure")
@@ -201,6 +202,28 @@ async def info(self, path, *, workdir):
         root = out.rstrip("\n")
         return {"url": url, "root": root}
 
+    async def get_plugins_update_date(self):
+        out = await self.read_lines(["svn", "ls", "-v", "^/tags", "http://plugins.svn.wordpress.org/"], ignore_errors=True)
+        line_pattern = re.compile("(?P<revision>\d+)\s+(?P<username>[\w\s\.-]+)\s+(?P<month>[A-Z][a-z]{2})\s+(?P<day>\d{2})\s+(?:(?P<year>\d{4})|(?P<time>\d\d:\d\d))\s+(?P<plugin>\S+)/$")
+        update_dates = {}
+        for line in out:
+            match = line_pattern.match(line)
+            if match:
+                plugin_key, day, month, year = match.group("plugin", "day", "month", "year")
+                if plugin_key is not ".":
+                    year = datetime.today().year if year is None else year
+                    date = datetime.strptime("%s %s %s" % (day, month, year), "%d %b %Y")
+                    update_dates[plugin_key] = date.date()
+        return update_dates
+
+    async def get_plugins_updated_since(self, date):
+        plugins_update_date = await self.get_plugins_update_date()
+        _plugins = set()
+        for key, update_date in plugins_update_date.items():
+            if update_date >= date:
+                _plugins.add(key)
+        return _plugins
+
     async def _process(self, command, workdir):
         process = await create_subprocess_exec(
             *command,

tests/common_test/vcs_test.py

@@ -18,7 +18,8 @@
 import asyncio
 from unittest import TestCase
 from unittest.mock import MagicMock, call, patch
-from fixtures import async_test, fake_future, file_path
+from fixtures import async_test, fake_future, file_path, freeze_time
+from datetime import date
 
 from openwebvulndb.common import RepositoryChecker, Repository
 from openwebvulndb.common.vcs import Subversion, SubversionWorkspace
@@ -325,6 +326,56 @@ async def test_svn_info(self, loop):
             self.assertEqual(info, {"url": "https://plugins.svn.wordpress.org/plugin/tags/1.0",
                                     "root": "https://plugins.svn.wordpress.org"})
 
+    @async_test()
+    async def test_svn_get_plugins_update_date_return_last_modification_date_of_tags_folder_for_plugins(self, loop):
+        with patch('openwebvulndb.common.vcs.create_subprocess_exec') as cse:
+            proc = MagicMock()
+            proc.stdout.readline.side_effect = [
+                fake_future(b"svn: warning: W160013: URL 'http://themes.svn.wordpress.org/tags' non-existent in revision 83065", loop=loop),
+                fake_future(b"1749964 user1               Oct 20 11:15 ./\n", loop=loop),
+                fake_future(b"1077807 user 2              Jan 28  2015 plugin-1/\n", loop=loop),
+                fake_future(b"1385952 user.3              Apr 04  2016 plugin-2/", loop=loop),
+                fake_future(b"svn: E200009: Could not list all targets because some targets don't exist", loop=loop)]
+            proc.stdout.at_eof.side_effect = [False, False, False, False, False, True]
+            proc.wait.return_value = fake_future(0, loop=loop)
+            cse.return_value = fake_future(proc, loop=loop)
+            svn = Subversion(loop=loop)
+
+            plugins = await svn.get_plugins_update_date()
+
+            cse.assert_has_calls([call(*("svn", "ls", "-v", "^/tags", "http://plugins.svn.wordpress.org/"), loop=loop,
+                                       stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL,
+                                       stdin=asyncio.subprocess.DEVNULL)])
+            self.assertEqual(plugins, {"plugin-1": date(year=2015, month=1, day=28),
+                                       "plugin-2": date(year=2016, month=4, day=4)})
+
+    @async_test()
+    async def test_svn_get_plugins_update_date_replace_hours_with_current_year(self, loop):
+        with patch('openwebvulndb.common.vcs.create_subprocess_exec') as cse:
+            proc = MagicMock()
+            proc.stdout.readline.return_value = \
+                fake_future(b"1749964 user1               Oct 20 11:15 plugin-1/\n", loop=loop)
+            proc.stdout.at_eof.side_effect = [False, True]
+            proc.wait.return_value = fake_future(0, loop=loop)
+            cse.return_value = fake_future(proc, loop=loop)
+            svn = Subversion(loop=loop)
+
+            plugins = await svn.get_plugins_update_date()
+
+            self.assertEqual(plugins, {"plugin-1": date(year=date.today().year, month=10, day=20)})
+
+    @freeze_time(date(year=2017, day=22, month=10))
+    @async_test()
+    async def test_svn_get_plugins_updated_since(self, loop):
+        plugins = {"plugin-0": date(year=2017, month=10, day=20), "plugin-1": date(year=2016, month=4, day=4),
+                   "plugin-2": date(year=2015, month=10, day=21), "plugin-3": date(year=2017, month=10, day=6)}
+        svn = Subversion(loop=None)
+        svn.get_plugins_update_date = MagicMock(return_value=fake_future(plugins, loop=loop))
+
+        recently_updated = await svn.get_plugins_updated_since(date(year=2017, day=6, month=10))
+
+        self.assertEqual(recently_updated, {"plugin-0", "plugin-3"})
+
 
 class SubversionWorkspaceTest(TestCase):