-
-
Save ghstwhl/5954d76fdd578324d954602106408be5 to your computer and use it in GitHub Desktop.
script to convert visz to ovpn - python3 viscosity-to-openvpn.py file.visz
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
# -*- coding: utf-8 -*- | |
"""Converts viscosity export files into an open vpn package | |
Usage: viscosity-to-openvpn.py <input> <output> | |
""" | |
import io | |
import os | |
import sys | |
import tarfile | |
import click | |
if sys.version.startswith('3'): | |
unicode = str | |
# ---------------------------------------------------------------------- | |
# Exceptions | |
# ---------------------------------------------------------------------- | |
class ConversionError(Exception): | |
"""Base conversion error""" | |
pass | |
class NoConnectionName(ConversionError): | |
"""No connection name was available""" | |
pass | |
class NoCertificateData(ConversionError): | |
"""No certificate data was found within certificate file""" | |
pass | |
class NoCertificateFile(ConversionError): | |
"""File was not available within archive""" | |
pass | |
# ---------------------------------------------------------------------- | |
# Command-line Interface | |
# ---------------------------------------------------------------------- | |
@click.command() | |
@click.argument('input-path', type=click.Path(exists=True)) | |
@click.argument('output', required=False, type=click.Path(), default=None) | |
def convert(input_path, output=None): | |
'''Converts Viscosity package | |
Args: | |
input (str): path to folder or file input | |
output (str): path to folder output [default: None] | |
''' | |
if input_path.endswith('.visc'): | |
output = input_path if output is None else output | |
if output and not os.path.exists(output): | |
output = input_path | |
files = [os.path.join(input_path, filename) for filename in os.listdir(input_path)] | |
for config_fp in files: | |
new_config = [] | |
if config_fp.endswith('.conf'): | |
with io.open(config_fp, encoding='utf-8') as stream: | |
connection_name = extract(stream, new_config, input_path=input_path) | |
new_config.insert(0, '# OpenVPN Config for {}'.format(connection_name)) | |
new_config = '\n'.join(new_config) + '\n' | |
output_filepath = os.path.join(output, '{}.ovpn'.format(connection_name)) | |
with io.open(output_filepath, 'w', encoding='utf-8') as out: | |
out.write(unicode(new_config)) | |
print('Wrote: {}'.format(output_filepath)) | |
elif input_path.endswith('.visz'): | |
if output is None: | |
output = os.path.dirname(input_path) | |
data = {} | |
with tarfile.open(input_path) as zipped: | |
for filepath, fileinfo in zip(zipped.getnames(), zipped.getmembers()): | |
if not fileinfo.isfile(): | |
continue | |
filename = filepath.split(os.path.sep)[-1] | |
data[filename] = zipped.extractfile(filepath).read() | |
for key in data: | |
if not key.endswith('.conf') or key.startswith('.'): | |
continue | |
new_config = [] | |
lines = data[key].decode("utf-8").split('\n') | |
connection_name = extract(lines, new_config, file_data=data) | |
new_config.insert(0, '# OpenVPN Config for {}'.format(connection_name)) | |
new_config = '\n'.join(new_config) + '\n' | |
output_filepath = os.path.join(output, '{}.ovpn'.format(connection_name)) | |
with io.open(output_filepath, 'w', encoding='utf-8') as out: | |
out.write(unicode(new_config)) | |
print('Wrote: {}'.format(output_filepath)) | |
# ---------------------------------------------------------------------- | |
# CLI Support | |
# ---------------------------------------------------------------------- | |
def extract(data, new_config, input_path=None, file_data={}): | |
certificate_files = ['ca', 'cert', 'key', 'tls-auth'] | |
connection_name = '' | |
for line in data: | |
line = line.rstrip() | |
if not line.strip(): | |
continue | |
# This was an invalid configuration, for some reason | |
elif line == 'compress lzo': | |
continue | |
elif line.startswith('#'): | |
if line.startswith('#viscosity name'): | |
connection_name = line.split('#viscosity name ', 1)[-1] | |
connection_name = connection_name.strip() | |
continue | |
try: | |
key, value = line.split(' ', 1) | |
value = value.strip() | |
except ValueError: | |
key, value = line, '' | |
if key in certificate_files: | |
if key == 'tls-auth': | |
try: | |
value, direction = value.split(' ', 1) | |
new_config.append('key-direction {}'.format(direction)) | |
except ValueError: | |
pass | |
if input_path: | |
cert_filepath = os.path.join(input_path, value) | |
with io.open(cert_filepath, encoding='utf-8') as cf: | |
certificate = cf.read() | |
else: | |
if value not in file_data: | |
raise NoCertificateFile('Could not find certificate file in archive') | |
certificate = file_data.get(value) | |
if not certificate: | |
raise NoCertificateData('Could not find certificate data') | |
new_config.append('<%s>' % key) | |
new_config.append(certificate.decode("utf-8") ) | |
new_config.append('</%s>' % key) | |
continue | |
new_config.append(line) | |
if not connection_name.strip(): | |
raise NoConnectionName('Could not find connection name in file. Aborting') | |
return connection_name | |
if __name__ == '__main__': | |
convert() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment