From 353c0099ee6903dd9f46a67c7d1f68b83f15de30 Mon Sep 17 00:00:00 2001 From: jwijenbergh Date: Tue, 29 Oct 2024 14:17:26 +0100 Subject: [PATCH] OpenVPN: Remove parser It was previously needed as we had to replace some stuff in the config IIRC. This is no longer needed, so instead of parsing and then recombining the config, let's pass it around as a string --- eduvpn/connection.py | 11 ++--- eduvpn/nm.py | 9 ++-- eduvpn/ovpn.py | 99 -------------------------------------------- eduvpn/storage.py | 12 ------ tests/test_nm.py | 7 +--- 5 files changed, 10 insertions(+), 128 deletions(-) delete mode 100644 eduvpn/ovpn.py diff --git a/eduvpn/connection.py b/eduvpn/connection.py index 23f9caf7..0b4f60c5 100644 --- a/eduvpn/connection.py +++ b/eduvpn/connection.py @@ -4,8 +4,6 @@ from enum import IntEnum from typing import Any, Dict, List, Optional -from eduvpn.ovpn import Ovpn - class Token: """The class that represents oauth Tokens @@ -172,14 +170,13 @@ def connect(self, callback): class OpenVPNConnection(Connection): - def __init__(self, ovpn: Ovpn): - self.ovpn = ovpn + def __init__(self, config_str): + self.config_str = config_str super().__init__() @classmethod def parse(cls, config_str: str) -> "OpenVPNConnection": # type: ignore - ovpn = Ovpn.parse(config_str) - return cls(ovpn=ovpn) + return cls(config_str=config_str) def connect( self, @@ -191,7 +188,7 @@ def connect( callback, ): manager.start_openvpn_connection( - self.ovpn, + self.config_str, default_gateway, dns_search_domains, callback=callback, diff --git a/eduvpn/nm.py b/eduvpn/nm.py index ac0a37d9..0d71f6e2 100644 --- a/eduvpn/nm.py +++ b/eduvpn/nm.py @@ -15,8 +15,7 @@ from eduvpn_common.main import EduVPN, Jar from gi.repository.Gio import Cancellable, Task # type: ignore -from eduvpn.ovpn import Ovpn -from eduvpn.storage import get_uuid, set_uuid, write_ovpn +from eduvpn.storage import get_uuid, set_uuid from eduvpn.utils import run_in_background_thread, run_in_glib_thread from eduvpn.variants import ApplicationVariant @@ -348,7 +347,7 @@ def ovpn_import(self, target: Path) -> Optional["NM.Connection"]: conn.normalize() return conn - def import_ovpn(self, ovpn: Ovpn) -> "NM.SimpleConnection": + def import_ovpn(self, ovpn: str) -> "NM.SimpleConnection": """ Import the OVPN string into Network Manager. """ @@ -356,7 +355,7 @@ def import_ovpn(self, ovpn: Ovpn) -> "NM.SimpleConnection": target = target_parent / f"{self.variant.name}.ovpn" _logger.debug(f"Writing configuration to {target}") with open(target, mode="w+t") as f: - ovpn.write(f) + f.write(ovpn) connection = self.ovpn_import(target) rmtree(target_parent) return connection @@ -401,7 +400,7 @@ def set_setting_ensure_permissions(self, con: "NM.SimpleConnection") -> "NM.Simp con.add_setting(s_con) return con - def start_openvpn_connection(self, ovpn: Ovpn, default_gateway, dns_search_domains, *, callback=None) -> None: + def start_openvpn_connection(self, ovpn: str, default_gateway, dns_search_domains, *, callback=None) -> None: _logger.debug("writing ovpn configuration to Network Manager") new_con = self.import_ovpn(ovpn) s_ip4 = new_con.get_setting_ip4_config() diff --git a/eduvpn/ovpn.py b/eduvpn/ovpn.py deleted file mode 100644 index f9cae47c..00000000 --- a/eduvpn/ovpn.py +++ /dev/null @@ -1,99 +0,0 @@ -from io import StringIO, TextIOWrapper -from typing import Iterable, List - - -class Item: - def to_string(self) -> str: - raise NotImplementedError - - def write(self, file: TextIOWrapper) -> None: - line = self.to_string() - file.write(f"{line}\n") - - def __eq__(self, other): - return self.__class__ is other.__class__ and self.__dict__ == other.__dict__ - - def __repr__(self): - fields = ",".join(f" {k}={v!r}" for k, v in self.__dict__.items()) - return f"<{self.__class__.__name__}{fields}>" - - -class Field(Item): - def __init__(self, name: str, arguments: List[str]) -> None: - self.name = name - self.arguments = arguments - - def to_string(self) -> str: - return f'{self.name} {" ".join(self.arguments)}' - - -class Section(Item): - def __init__(self, tag: str, content: List[str]) -> None: - self.tag = tag - self.content = content - - def to_string(self) -> str: - lines = [] - lines.append(f"<{self.tag}>") - lines.extend(self.content) - lines.append(f"") - return "\n".join(lines) - - -class Comment(Item): - def __init__(self, content: str) -> None: - self.content = content - - def to_string(self) -> str: - return f"#{self.content}" - - -class Empty(Item): - def to_string(self): - return "" - - -class InvalidOVPN(Exception): - pass - - -def parse_ovpn(lines: Iterable[str]) -> Iterable[Item]: - current_section = None - for _lineno, line in enumerate(lines): - if current_section is not None: - if line.strip() == f"": - yield current_section - current_section = None - else: - current_section.content.append(line) - elif line.startswith("#"): - yield Comment(line.rstrip()[1:]) - elif line.startswith("<"): - line = line.rstrip() - assert line.endswith(">") - section = Section(line[1:-1], []) - current_section = section - elif line.rstrip() == "": - yield Empty() - else: - field_name, *arguments = line.rstrip().split() - yield Field(field_name, arguments) - assert current_section is None - - -class Ovpn: - def __init__(self, content: List[Item]) -> None: - self.content = content - - @classmethod - def parse(cls, content: str) -> "Ovpn": - return cls(list(parse_ovpn(content.splitlines()))) - - def write(self, file: TextIOWrapper) -> None: - for item in self.content: - item.write(file) - - def to_string(self) -> str: - file = StringIO() - self.write(file) - return file.getvalue() diff --git a/eduvpn/storage.py b/eduvpn/storage.py index 5e1ba41c..8235b862 100644 --- a/eduvpn/storage.py +++ b/eduvpn/storage.py @@ -4,7 +4,6 @@ from typing import Optional -from eduvpn.ovpn import Ovpn from eduvpn.settings import CONFIG_DIR_MODE, CONFIG_PREFIX from eduvpn.utils import get_logger @@ -45,17 +44,6 @@ def set_setting(variant, what: str, value: str): f.write(value) -def write_ovpn(ovpn: Ovpn, private_key: str, certificate: str, target: PathLike): - """ - Write the OVPN configuration file to target. - """ - logger.info(f"Writing configuration to {target}") - with open(target, mode="w+t") as f: - ovpn.write(f) - f.writelines(f"\n\n{private_key}\n\n") - f.writelines(f"\n\n{certificate}\n\n") - - def get_uuid(variant) -> Optional[str]: """ Read the UUID of the last generated eduVPN Network Manager connection. diff --git a/tests/test_nm.py b/tests/test_nm.py index 3adb4455..a93251bb 100644 --- a/tests/test_nm.py +++ b/tests/test_nm.py @@ -1,7 +1,6 @@ from unittest import TestCase, skipIf from eduvpn.nm import NMManager -from eduvpn.ovpn import Ovpn from eduvpn.variants import EDUVPN from tests.mock_config import mock_config @@ -14,13 +13,11 @@ def test_nm_available(self): def test_import_ovpn(self): nm_manager = NMManager(EDUVPN, None) - ovpn = Ovpn.parse(mock_config) - nm_manager.import_ovpn(ovpn) + nm_manager.import_ovpn(mock_config) def test_get_add_connection(self): nm_manager = NMManager(EDUVPN, None) - ovpn = Ovpn.parse(mock_config) - simple_connection = nm_manager.import_ovpn(ovpn) + simple_connection = nm_manager.import_ovpn(mock_config) nm_manager.add_connection(simple_connection) def test_get_uuid(self):